Шпаргалка по параллелизму в Python: потоки, процессы, asyncio

Python предлагает три подхода к параллелизму: потоки (threading — I/O-bound), процессы (multiprocessing — CPU-bound) и асинхронность (asyncio — I/O-bound, высокая производительность). Начиная с Python 3.13 (экспериментально) и 3.14 (официально) доступен free-threaded режим — CPython без GIL с настоящим параллелизмом потоков.

GIL и выбор технологии

GIL (Global Interpreter Lock) — мьютекс, разрешающий только одному потоку исполнять Python-код. Поэтому:

Threading — потоки

Потоки разделяют память процесса. GIL не даёт параллелиться CPU-задачам, но для I/O (сеть, файлы) — отлично.

Создание и жизненный цикл потока

import threading
import time

# Простой поток
def worker(name, delay):
    print(f"[{name}] Старт")
    time.sleep(delay)
    print(f"[{name}] Финиш")

t = threading.Thread(target=worker, args=("A", 2), name="Worker-A", daemon=False)
t.start()        # запустить
t.is_alive()     # True — поток выполняется
t.ident          # уникальный ID потока (int)
t.native_id      # системный ID потока (OS-level)
t.name           # 'Worker-A'
t.daemon         # False — главный процесс ждёт завершения

t.join()         # ждать завершения
t.join(timeout=5.0)  # ждать не более 5 секунд
t.is_alive()     # False — поток завершён

# Даймон-потоки — не блокируют завершение программы
# Когда главный поток выходит, все daemon-потоки убиваются
t = threading.Thread(target=worker, daemon=True)
t.start()
# программа НЕ будет ждать t — выйдет сразу после главного потока

Главный поток и служебные функции

# Главный поток — тот, в котором запущен интерпретатор
main = threading.main_thread()         # объект главного потока
main.name                              # 'MainThread'

# Служебные функции
threading.current_thread()             # текущий поток (объект)
threading.active_count()               # число живых потоков (включая главный)
threading.enumerate()                  # список всех живых потоков

# Пример: найти все потоки кроме главного
main = threading.main_thread()
for t in threading.enumerate():
    if t is not main:
        print(f"{t.name}: is_alive={t.is_alive()}, daemon={t.daemon}")

Хранение локальных данных потока — threading.local()

Потоки разделяют глобальные переменные. threading.local() — контейнер, у которого каждый поток видит свои значения. Потокобезопасно без блокировок.

import threading

# БЕЗ local — гонка данных
# data = {}  # все потоки пишут в один словарь

# С local — у каждого потока своя копия
thread_data = threading.local()

def process(request_id):
    thread_data.request_id = request_id      # видит только этот поток
    thread_data.user = fetch_user(request_id)
    save_to_db(thread_data.user)              # не надо таскать параметры!

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

Атомарные операции под GIL

Под GIL некоторые операции выглядят атомарными, но полагаться на это нельзя.

# Что условно-атомарно под GIL (но не в free-threading!):
# - чтение/запись одной переменной
# - list.append(), list.pop(), dict[key] = value
# - операции с локальными переменными (не global!)

# Что НЕ атомарно даже под GIL:
counter = 0
def increment():
    global counter
    counter += 1        # чтение → сложение → запись = 3 шага!
                        # Другой поток может вклиниться между ними.

# Всегда используй Lock для изменяемого shared state!

ThreadPoolExecutor — пул потоков

from concurrent.futures import ThreadPoolExecutor, as_completed, wait
import concurrent.futures

def fetch(url):
    return f"Данные с {url}"

urls = ["url1", "url2", "url3", "url4"]

with ThreadPoolExecutor(max_workers=4) as executor:
    # 1. map — сохраняет порядок, блокируется до завершения всех
    results = executor.map(fetch, urls)
    for r in results:
        print(r)

    # 2. submit — возвращает Future, можно обрабатывать по мере готовности
    futures = [executor.submit(fetch, url) for url in urls]

    # as_completed — обрабатывать результаты по мере завершения (порядок не гарантирован)
    for future in as_completed(futures):
        result = future.result()
        print(f"Готово: {result}")

    # 3. wait — ждать завершения с гибкими условиями
    done, not_done = wait(futures, return_when=concurrent.futures.FIRST_COMPLETED)
    for f in done:
        print(f.result())  # обработать первый готовый

    done, _ = wait(futures, return_when=concurrent.futures.FIRST_EXCEPTION)
    # done — все, что завершились до ПЕРВОГО исключения

    # return_when варианты:
    # FIRST_COMPLETED  — ждать первого завершившегося
    # FIRST_EXCEPTION   — ждать первого исключения (или всех)
    # ALL_COMPLETED     — ждать всех (по умолчанию)

# shutdown — закрыть пул, дождаться задач (вызывается автоматически в with)
with ThreadPoolExecutor(max_workers=4) as executor:
    futures = [executor.submit(fetch, url) for url in urls]
    executor.shutdown(wait=True, cancel_futures=False)
    # wait=True         — ждать завершения выполняющихся задач
    # cancel_futures=True — отменить ожидающие задачи (Python 3.9+)

# initializer — функция, запускаемая в каждом потоке при старте
def init_thread():
    threading.local().db = connect_db()

with ThreadPoolExecutor(max_workers=4, initializer=init_thread) as executor:
    executor.map(process, data)

Future — объект будущего результата

future = executor.submit(fetch, "url")

future.done()          # True — задача завершена (успех/ошибка/отмена)
future.cancelled()     # True — задача была отменена
future.running()       # True — задача выполняется прямо сейчас

result = future.result(timeout=10)    # получить результат (блокируется)
future.exception(timeout=10)          # получить исключение (или None)

# Через какое время вернуть результат по умолчанию (без таймаута)
future.result()  # ждёт бесконечно

# add_done_callback — вызвать функцию, когда задача завершится
def on_done(future):
    try:
        result = future.result()
        print(f"Успех: {result}")
    except Exception as e:
        print(f"Ошибка: {e}")

future.add_done_callback(on_done)

ThreadPoolExecutor vs asyncio

# Когда НЕЛЬЗЯ asyncio — используй ThreadPoolExecutor:
with ThreadPoolExecutor(max_workers=10) as executor:
    results = list(executor.map(requests.get, urls))  # requests — синхронный!

# Когда можно asyncio — всегда предпочитай его:
async with aiohttp.ClientSession() as session:
    tasks = [fetch(session, url) for url in urls]
    results = await asyncio.gather(*tasks)

Примитивы синхронизации

import threading

# Lock — простая блокировка
lock = threading.Lock()
counter = 0

def increment():
    global counter
    with lock:           # acquire/release через контекстный менеджер
        counter += 1

# Lock вручную
lock.acquire()
# ... критическая секция ...
lock.release()

# RLock — рекурсивная (можно захватывать повторно из того же потока)
rlock = threading.RLock()
rlock.acquire()
rlock.acquire()  # OK — тот же поток
rlock.release()
rlock.release()

# Semaphore — ограничение N одновременных доступов
sem = threading.Semaphore(3)  # не более 3 потоков одновременно
with sem:
    ...  # дорогая операция

# Event — сигнал между потоками
event = threading.Event()

def waiter():
    print("Жду сигнала...")
    event.wait()              # блокируется до event.set()
    print("Получил сигнал!")

def signaller():
    time.sleep(2)
    event.set()               # разбудить всех ждущих
    event.clear()             # сбросить флаг

# Condition — ожидание условия
condition = threading.Condition()
items = []

def consumer():
    with condition:
        while not items:
            condition.wait()      # ждать, пока producer не вызовет notify
        item = items.pop()
        print(f"Получил: {item}")

def producer():
    with condition:
        items.append("data")
        condition.notify()        # разбудить одного consumer
        # condition.notify_all()  # разбудить всех

# Barrier — ждать, пока соберутся все N потоков
barrier = threading.Barrier(3)

def worker():
    print("Фаза 1 завершена, жду остальных...")
    barrier.wait()                # блокируется, пока все 3 не дойдут
    print("Все готовы, продолжаем!")

# Timer — отложенный запуск
timer = threading.Timer(5.0, worker, args=("delayed",))
timer.start()
timer.cancel()  # отменить, если нужно

Thread-safe очередь

import queue

q = queue.Queue(maxsize=10)        # FIFO
q = queue.LifoQueue()              # LIFO (стек)
q = queue.PriorityQueue()          # приоритетная

# Producer-consumer
def producer(q):
    for i in range(5):
        q.put(f"item-{i}")
        print(f"Произвёл item-{i}")

def consumer(q):
    while True:
        item = q.get()
        print(f"Потребил {item}")
        q.task_done()
        if item == "item-4": break

# q.get() блокируется, если очередь пуста
# q.get(timeout=2) — таймаут
# q.put(item, timeout=2) — таймаут если очередь полна

Multiprocessing — процессы

Процессы имеют собственный GIL и память. Полный параллелизм на всех ядрах. Данные передаются через pickle (медленно для больших объёмов).

Создание, атрибуты и управление процессами

import multiprocessing as mp

def cpu_heavy(n):
    return sum(i*i for i in range(n))

# Создание и запуск
p = mp.Process(target=cpu_heavy, args=(10_000_000,), name="Worker-1")
p.start()

# Атрибуты процесса
p.pid           # PID процесса (int) или None, если не запущен
p.name          # 'Worker-1'
p.is_alive()    # True — процесс выполняется
p.exitcode      # код возврата: None (ещё работает), 0 (успех), >0 (ошибка), <0 (убит сигналом)
p.daemon        # daemon-процессы убиваются при выходе главного

p.join()        # ждать завершения
p.join(timeout=5.0)

# Завершение процесса
p.terminate()    # SIGTERM — вежливая просьба завершиться
p.kill()         # SIGKILL — принудительное убийство (сразу, без очистки)
p.close()        # освободить ресурсы (после join или kill)
# После close() нельзя вызывать start/join снова

# Перезапуск процесса — создать новый объект
# p.start()  # ❌ AssertionError — процесс уже использован
p2 = mp.Process(target=cpu_heavy, args=(10_000_000,))
p2.start()

Главный процесс и служебные функции

mp.current_process()         # текущий процесс (объект)
mp.active_children()         # список живых дочерних процессов
mp.cpu_count()               # число логических ядер CPU
mp.parent_process()          # родительский процесс (Python 3.8+)

# Идиома: защита точки входа (обязательно для spawn-метода на Windows/macOS!)
if __name__ == '__main__':
    p = mp.Process(target=worker)
    p.start()
    p.join()
# Без этой защиты — бесконечная рекурсия создания процессов!

Обмен данными между процессами

import multiprocessing as mp

# Queue — основная потокобезопасная очередь
q = mp.Queue(maxsize=10)
q.put("item")
item = q.get()
q.empty(), q.full(), q.qsize()  # приблизительные значения!

# SimpleQueue — упрощённая (нет maxsize, быстрее)
sq = mp.SimpleQueue()
sq.put("item")
item = sq.get()

# JoinableQueue — очередь с отслеживанием завершения задач
jq = mp.JoinableQueue()

def worker(q):
    while True:
        item = q.get()
        if item is None: break
        process(item)
        q.task_done()              # отмечаем: задача обработана

jq.put("task1"); jq.put("task2")
workers = [mp.Process(target=worker, args=(jq,)) for _ in range(2)]
for w in workers: w.start()

jq.join()         # ждать, пока все task_done() будут вызваны
for _ in workers: jq.put(None)  # сигнал остановки

# Pipe — двусторонний канал (только между двумя процессами!)
parent_conn, child_conn = mp.Pipe()      # duplex=True (двусторонний)
# parent_conn, child_conn = mp.Pipe(duplex=False)  # однонаправленный

def child(conn):
    conn.send("Привет от child")
    msg = conn.recv()
    conn.close()

p = mp.Process(target=child, args=(child_conn,))
p.start()
msg = parent_conn.recv()
parent_conn.send("Привет от parent")
p.join()

# Value и Array — разделяемая память (без копирования через pickle)
counter = mp.Value('i', 0)              # 'i'=int, 'd'=double, 'b'=signed char
arr = mp.Array('i', [0]*10)             # 'f'=float, 'c'=char, 'l'=long
flag = mp.Value('b', False)             # 'b' для bool

with counter.get_lock():
    counter.value += 1

# Без блокировки — быстрее, но небезопасно при одновременной записи
counter.value += 1

# Manager — прокси-объекты (медленнее, но удобнее — любые типы!)
with mp.Manager() as manager:
    shared_list = manager.list([1, 2, 3])
    shared_dict = manager.dict({'a': 1})
    shared_set = manager.set()           # множество
    shared_queue = manager.Queue()       # прокси-очередь
    shared_lock = manager.Lock()         # прокси-блокировка
    ns = manager.Namespace()             # объект с произвольными атрибутами
    ns.x = 42; ns.name = "shared"

    # Manager создаёт серверный процесс — все операции через прокси (медленно!)
    shared_dict['key'] = 'value'         # прокси-вызов через IPC

ProcessPoolExecutor — пул процессов

from concurrent.futures import ProcessPoolExecutor, as_completed
import os

def process(x):
    return f"PID {os.getpid()}: {x ** 2}"

# initializer — функция при старте каждого процесса (подключение к БД и т.д.)
def init_worker():
    import signal
    signal.signal(signal.SIGINT, signal.SIG_IGN)  # игнорировать Ctrl+C в воркерах

with ProcessPoolExecutor(max_workers=4, initializer=init_worker) as executor:
    # map — сохраняет порядок
    results = list(executor.map(process, range(100)))

    # submit + as_completed — обрабатывать по мере готовности
    futures = [executor.submit(process, x) for x in range(100)]
    for future in as_completed(futures):
        result = future.result()

    # shutdown (вызывается автоматически в with)
    executor.shutdown(wait=True, cancel_futures=False)

# Pool (старый API, всё ещё встречается)
with mp.Pool(processes=4) as pool:
    results = pool.map(process, range(100))         # блокируется, порядок сохранён
    for r in pool.imap(process, range(100)):        # ленивый, порядок сохранён
        print(r)
    for r in pool.imap_unordered(process, range(100)):  # ленивый, порядок НЕ сохранён
        print(r)
    result = pool.apply_async(process, (42,))       # асинхронный вызов одной задачи
    print(result.get())

Asyncio — асинхронность

Однопоточный event loop. Идеален для тысяч одновременных I/O-операций. async/await — корутины, asyncio — цикл событий.

Базовые концепции и запуск

import asyncio

# async def — корутина (awaitable объект)
async def greet(name, delay):
    await asyncio.sleep(delay)    # НЕ time.sleep!
    return f"Привет, {name}!"

# asyncio.run() — точка входа. Создаёт event loop, ждёт корутину, закрывает loop.
asyncio.run(greet("Мир", 1))

# Что делает asyncio.run() под капотом:
# loop = asyncio.new_event_loop()
# asyncio.set_event_loop(loop)
# try:
#     return loop.run_until_complete(coro)
# finally:
#     loop.close()

Задачи и конкурентное выполнение

async def main():
    # create_task — планирует корутину на выполнение (НЕ ждёт результат)
    task1 = asyncio.create_task(greet("Анна", 2))
    task2 = asyncio.create_task(greet("Борис", 1))

    # Обе корутины уже запущены! await ждёт результат.
    r1 = await task1
    r2 = await task2

    # gather — запустить все и ждать все (возвращает список результатов)
    results = await asyncio.gather(
        greet("Анна", 1),
        greet("Борис", 2),
        # return_exceptions=True — исключения возвращаются вместо выбрасывания
    )

    # TaskGroup (Python 3.11+) — gather с защитой от утечек задач
    async with asyncio.TaskGroup() as tg:
        tg.create_task(greet("Анна", 1))
        tg.create_task(greet("Борис", 2))
    # одна упала → все остальные отменяются

asyncio.run(main())

Таймауты и ожидание по условиям

# wait_for — таймаут для одной корутины
async def main():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=5.0)
    except asyncio.TimeoutError:
        print("Операция не уложилась в 5 секунд")

    # Python 3.11+: asyncio.timeout() — контекстный менеджер
    try:
        async with asyncio.timeout(5.0):
            result = await slow_operation()
    except TimeoutError:
        print("Таймаут!")

# wait — гибкое ожидание нескольких задач
async def main():
    tasks = [asyncio.create_task(greet(name, i)) for i, name in enumerate(["A","B","C"])]

    # Ждать первую завершившуюся
    done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
    for task in done:
        print(task.result())   # только первая

    # Отменить оставшиеся
    for task in pending:
        task.cancel()

    # return_when варианты:
    # asyncio.FIRST_COMPLETED   — ждать первую завершившуюся
    # asyncio.FIRST_EXCEPTION   — ждать первую с ошибкой (или все)
    # asyncio.ALL_COMPLETED     — ждать все (по умолчанию)

# as_completed — итератор по мере готовности
async def main():
    tasks = [asyncio.create_task(greet(name, delay)) for name, delay in [("A",3),("B",1),("C",2)]]
    for coro in asyncio.as_completed(tasks):
        result = await coro        # получаем первый готовый результат, потом второй...
        print(f"Готово: {result}")
    # Порядок: B → C → A (по задержке, не по порядку создания!)

Управление задачами: отмена, проверка статуса, колбэки

async def main():
    task = asyncio.create_task(slow_operation())

    # Статус задачи
    task.done()              # True — завершена (успех, ошибка или отмена)
    task.cancelled()         # True — задача была отменена
    if not task.done():
        ...

    result = task.result()   # получить результат (если done)
    exc = task.exception()   # получить исключение (если done с ошибкой)

    # Отмена задачи
    task.cancel()            # бросает CancelledError в корутину
    try:
        result = await task
    except asyncio.CancelledError:
        print("Задача отменена")

    # Колбэк при завершении (вызывается независимо от результата)
    def on_done(task):
        if task.cancelled():
            print("Отменена")
        elif task.exception():
            print(f"Ошибка: {task.exception()}")
        else:
            print(f"Успех: {task.result()}")

    task.add_done_callback(on_done)
    # add_done_callback можно добавить несколько — выполнятся по порядку

    # shield — защита от cancel (родитель отменён, но эта задача — нет)
    result = await asyncio.shield(critical_operation())
    # CancelledError НЕ пробросится в critical_operation

    # ensure_future — старый способ создать Task (create_task предпочтительнее)
    task = asyncio.ensure_future(coro)      # = create_task

Event loop — жизненный цикл

# Получить текущий event loop
loop = asyncio.get_running_loop()    # внутри async-функции (строго!)
# loop = asyncio.get_event_loop()     # вне async-функции (deprecated поведение в 3.10+)

# Создать новый loop (для низкоуровневого использования)
new_loop = asyncio.new_event_loop()
asyncio.set_event_loop(new_loop)     # установить как текущий
try:
    new_loop.run_until_complete(coro)
finally:
    new_loop.close()

# Проверить, запущен ли loop
asyncio.get_event_loop().is_running()
asyncio.get_event_loop().is_closed()

# Политика event loop (менять редко, для продвинутых случаев)
asyncio.get_event_loop_policy()
# asyncio.WindowsSelectorEventLoopPolicy()  # для Windows (по умолчанию Proactor)
# asyncio.DefaultEventLoopPolicy()          # Unix

Асинхронные конструкции: async for, async with, async генераторы

import asyncio

# async with — асинхронный контекстный менеджер
class AsyncConnection:
    async def __aenter__(self):
        await asyncio.sleep(0.1)      # асинхронное подключение
        return self

    async def __aexit__(self, *args):
        await asyncio.sleep(0.1)      # асинхронное отключение

async def main():
    async with AsyncConnection() as conn:
        ...  # работа с conn

    # aiofiles — пример из реальной жизни
    import aiofiles
    async with aiofiles.open('file.txt', 'r') as f:
        content = await f.read()

# async for — асинхронный итератор
class AsyncRange:
    def __init__(self, n):
        self.n = n
        self.i = 0

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.i >= self.n:
            raise StopAsyncIteration
        await asyncio.sleep(0.1)      # асинхронная пауза между элементами
        self.i += 1
        return self.i

async def main():
    async for i in AsyncRange(5):
        print(i)

    # aiofiles: асинхронная построчная обработка
    async with aiofiles.open('file.txt', 'r') as f:
        async for line in f:
            await process(line)

# Асинхронный генератор (yield в async-функции)
async def generate_numbers(n):
    for i in range(n):
        await asyncio.sleep(0.1)
        yield i

async def main():
    async for num in generate_numbers(5):
        print(num)

    # Асинхронные list/dict/set comprehensions (Python 3.6+)
    results = [x async for x in generate_numbers(5)]
    # Требует async for — работает только внутри async-функции

Примитивы синхронизации asyncio

import asyncio

# Lock
lock = asyncio.Lock()
async with lock:
    ...  # критическая секция

# Semaphore — ограничение N одновременных
sem = asyncio.Semaphore(5)
async with sem:
    await do_request()

# BoundedSemaphore — как Semaphore, но с защитой от лишних release()
sem = asyncio.BoundedSemaphore(5)

# Event
event = asyncio.Event()
await event.wait()      # ждать сигнала
event.set()             # подать сигнал (разблокировать всех)
event.clear()           # сбросить
event.is_set()          # True/False

# Condition
cond = asyncio.Condition()
async with cond:
    await cond.wait()           # ждать notify
    await cond.wait_for(predicate)  # ждать, пока predicate() == True
    cond.notify(n=1)            # разбудить n ждущих
    cond.notify_all()           # разбудить всех

# Queue
q = asyncio.Queue(maxsize=10)
await q.put(item)
item = await q.get()
q.task_done()                  # отметить: элемент обработан
await q.join()                 # ждать, пока все task_done()

q.qsize()                      # примерный размер (не потокобезопасный)
q.empty(), q.full()            # примерно

Синхронный код в asyncio

import asyncio
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=4)

def blocking_io():
    import time
    time.sleep(2)
    return "Готово"

async def main():
    loop = asyncio.get_running_loop()

    # Запуск синхронной функции в отдельном потоке (не блокирует event loop!)
    result = await loop.run_in_executor(executor, blocking_io)
    # Python 3.9+: более короткий синтаксис
    result = await asyncio.to_thread(blocking_io)

    # Для CPU-bound (если нужно именно из asyncio):
    result = await loop.run_in_executor(None, cpu_heavy)  # None = ProcessPoolExecutor по умолчанию

Сравнение и выбор

# I/O-bound: сеть, БД, файлы
#   → asyncio (сотни/тысячи соединений)
#   → threading (простая интеграция с синхронными библиотеками)

# CPU-bound: вычисления, обработка данных
#   → free-threaded Python 3.14+ (без GIL, настоящий параллелизм потоков)
#   → multiprocessing (если free-threaded недоступен или C-расширения несовместимы)

# Смешанные задачи
#   → multiprocessing + asyncio внутри каждого процесса

Python 3.14 Free-Threading (без GIL)

Начиная с Python 3.13 (экспериментально) и Python 3.14 (официально), CPython поддерживает free-threaded сборку — режим без GIL. Потоки работают по-настоящему параллельно на всех ядрах CPU.

Установка и проверка

# Windows/Mac: официальный инсталлятор → галочка "free-threaded binaries"
# Запуск:
python3.14t              # free-threaded версия

# Linux: сборка из исходников
./configure --disable-gil
make -j$(nproc)
sudo make install

# Проверка
python -VV               # "... (free-threading build)" — если free-threaded

import sys
import sysconfig

# Программная проверка
is_free_threaded = sysconfig.get_config_var("Py_GIL_DISABLED") == 1
gil_enabled = sys._is_gil_enabled()
print(f"Free-threaded: {is_free_threaded}, GIL: {gil_enabled}")

Производительность: до 4x на CPU-задачах

Тот же самый многопоточный код работает быстрее без изменений:

import threading, time

def cpu_task(n):
    total = 0
    for i in range(n):
        total += i * i
    return total

def bench():
    threads = [threading.Thread(target=cpu_task, args=(20_000_000,)) for _ in range(4)]
    start = time.time()
    for t in threads: t.start()
    for t in threads: t.join()
    return time.time() - start

# Стандартный Python 3.14 (с GIL):    ~5.8 сек (последовательно)
# Free-threaded Python 3.14t:         ~1.5 сек (4x быстрее!)

Управление GIL в рантайме

# Включить GIL для конкретного запуска
PYTHON_GIL=1 python3.14t script.py
python3.14t -X gil=1 script.py

# Выключить (по умолчанию в free-threaded сборке)
python3.14t -X gil=0 script.py

GIL автоматически включается, если импортируется C-расширение без поддержки free-threading. При этом выводится предупреждение. Статус совместимости пакетов: py-free-threading.github.io/tracking.

Что меняется в поведении

# 1. Контекстные переменные: потоки наследуют контекст родителя
# В free-threaded режиме thread_inherit_context=True по умолчанию

# 2. Встроенные типы (dict, list, set) используют внутренние блокировки
# Это защищает от крашей, НО не гарантирует логическую корректность!
# Используй явные threading.Lock() как и раньше.

# 3. Однонопоточный оверхед: +1-8% (зависит от платформы)
# macOS aarch64: ~1%, Linux x86-64: ~8%

# 4. Повышенное потребление памяти
# - Объекты без GC стали крупнее (None: 32 байта вместо 16)
# - mimalloc вместо pymalloc (другой аллокатор)
# - Отложенное освобождение памяти (QSBR)
# Вызов gc.collect() ускоряет возврат памяти ОС

Миграция старого многопоточного кода

# 1. НИЧЕГО не меняй, если код уже корректно синхронизирован
#    Весь threading.Lock / RLock / Semaphore продолжает работать.

# 2. Проверь код на race conditions — без GIL они проявляются чаще:
#    БЫЛО (GIL скрывал проблему):
counter = 0
def increment():
    global counter
    counter += 1        # НЕ атомарно! Без GIL — гонка данных

#    СТАЛО:
import threading
lock = threading.Lock()
def increment():
    global counter
    with lock:
        counter += 1    # ✅ потокобезопасно

# 3. Замени self-written спин-блокировки на threading-примитивы
#    Без GIL наивные while-циклы жрут CPU на всех ядрах.

# 4. Не полагайся на неявную потокобезопасность встроенных типов:
#    БЫЛО (работало под GIL, под free-threading — гонка):
data = {}
def add(key, value):
    data[key] = value   # dict защищён от краша, но данные могут потеряться

#    СТАЛО:
def add(key, value):
    with lock:
        data[key] = value  # ✅ корректно

# 5. Проверь C-расширения зависимостей:
#    pip list | while read pkg; do
#        python -c "import $pkg" 2>&1 | grep -q "GIL" && echo "$pkg: NO"
#    done

# 6. Протестируй под free-threaded сборкой:
#    PYTHON_GIL=0 python3.14t -m pytest tests/

# 7. Опционально: добавь явное отключение GIL в коде
import sysconfig
if sysconfig.get_config_var("Py_GIL_DISABLED"):
    print("Running free-threaded — используем потоки вместо процессов")
    from concurrent.futures import ThreadPoolExecutor as Executor
else:
    from concurrent.futures import ProcessPoolExecutor as Executor

Когда использовать free-threading

Примитивы и паттерны синхронизации

Mutex (Lock) — взаимное исключение

Самый базовый примитив. Только один поток может войти в критическую секцию. Все остальные ждут.

Когда использовать: любое изменение разделяемых данных. Если два потока пишут в одну переменную — нужен Lock.

Важно: Lock должен быть ОДИН на все потоки. Создавать Lock внутри функции-потока — ошибка (у каждого будет свой, никто никого не ждёт).

import threading

# Правильно: один Lock на всех
lock = threading.Lock()
counter = 0

def increment():
    global counter
    with lock:
        counter += 1    # чтение-изменение-запись защищены

# Неправильно: Lock внутри функции
def bad_increment():
    local_lock = threading.Lock()  # каждый поток создаёт СВОЙ Lock!
    with local_lock:
        counter += 1               # защиты нет — все пройдут одновременно

Вариант: try-lock (не ждать, если занято)

if lock.acquire(blocking=False):   # или lock.acquire(timeout=0)
    try:
        counter += 1
    finally:
        lock.release()
else:
    print("Занято другим потоком — пропускаем")

RLock — рекурсивный мьютекс

Тот же поток может захватить RLock несколько раз (счётчик захватов). Разблокировка — столько же release().

Когда использовать: функция A (с Lock) вызывает функцию B (тоже с Lock), и они могут быть вызваны из одного потока. Обычный Lock приведёт к deadlock.

rlock = threading.RLock()

def outer():
    with rlock:           # захват 1
        inner()

def inner():
    with rlock:           # захват 2 — OK, тот же поток
        counter += 1

Semaphore — ограничение количества одновременных доступов

Не «только один», а «не более N одновременно». Счётчик уменьшается при acquire, увеличивается при release. Если счётчик = 0 — acquire блокируется.

Когда использовать: ограничение числа одновременных соединений к БД, ограничение конкурентных HTTP-запросов, пул ресурсов.

import threading

# Не более 3 потоков одновременно в критической секции
db_semaphore = threading.Semaphore(3)

def query_db(query):
    with db_semaphore:
        result = execute_query(query)   # не более 3 одновременных запросов
        return result

# BoundedSemaphore — дополнительно проверяет, что release() не вызывается лишний раз
sem = threading.BoundedSemaphore(3)
sem.release()  # ValueError: Semaphore released too many times

Event — сигнал «готово» между потоками

Один поток ждёт сигнала (wait()), другой подаёт его (set()). Это одноразовый сигнал — после set() все текущие и будущие wait() проходят мгновенно.

Когда использовать: главный поток ждёт инициализации воркеров, координация фаз выполнения, сигнал остановки.

event = threading.Event()

def worker():
    event.wait()              # ждём сигнал
    print("Начинаю работу!")   # все воркеры стартуют одновременно

def controller():
    init_resources()
    event.set()                # сигнал всем воркерам

# Сброс флага (для повторного использования)
event.clear()                  # возвращает в неподанное состояние
event.is_set()                 # True — сигнал подан

Condition — ожидание сложного условия

Комбинация Lock + Event. Поток ждёт (wait()), пока другой не изменит состояние и не вызовет notify(). В отличие от Event, Condition проверяет предикат (условие), а не просто флаг.

Когда использовать: producer-consumer (производитель-потребитель), ожидание заполнения буфера, ожидание освобождения ресурса.

import threading

condition = threading.Condition()
buffer = []
MAX_SIZE = 5

def producer():
    for i in range(10):
        with condition:
            while len(buffer) >= MAX_SIZE:   # буфер полон — ждём
                condition.wait()
            buffer.append(i)
            condition.notify()                # будим consumer'а

def consumer():
    while True:
        with condition:
            while not buffer:                 # буфер пуст — ждём
                condition.wait()
            item = buffer.pop(0)
            condition.notify()                # будим producer'а
            if item == 9: return

# notify() будит ОДИН случайный ждущий поток
# notify_all() будит ВСЕ ждущие потоки

Barrier — синхронизация группы потоков

Все N потоков должны дойти до барьера, прежде чем кто-либо продолжит. Работает как «точка сбора».

Когда использовать: параллельные вычисления по фазам (все потоки завершили фазу 1 → начинаем фазу 2), одновременный старт.

import threading

barrier = threading.Barrier(3, timeout=5)    # 3 потока, таймаут 5с

def worker(phase):
    # Фаза 1
    do_phase1(phase)
    barrier.wait()                            # все завершили фазу 1
    # Фаза 2 (начинается только когда все прошли барьер)
    do_phase2(phase)
    barrier.wait()
    print("Все фазы завершены")

# barrier.wait() возвращает ID (0..N-1) — уникальный номер прошедшего
# barrier.reset() — сброс (если нужно переиспользовать)
# barrier.abort()  — принудительный срыв (все ждущие получат BrokenBarrierError)

Атомарные операции (atomics)

В Python нет встроенных атомарных операций как в C++/Rust. Под GIL отдельные операции выглядят атомарными, но полагаться на это нельзя (особенно в free-threading).

# Под GIL условно-атомарно (НО не гарантируется спецификацией!):
x = 42               # присваивание простой переменной
lst.append(item)     # один вызов метода (CPython выполняет за один опкод)
d[key] = value       # присваивание в словарь

# Никогда не атомарно даже под GIL:
x += 1               # чтение + сложение + запись = 3 шага
lst[0] += 1          # чтение элемента + сложение + запись
d[key] += 1          # то же самое

# В free-threading (Python 3.13+) ВООБЩЕ НИЧЕГО не атомарно без Lock!

# Правильный подход: всегда использовать явную синхронизацию
with lock:
    counter += 1        # ✅ гарантированно потокобезопасно

Read-Write Lock (много читателей, один писатель)

Python не имеет встроенного RWLock. Имитация — два Lock + счётчик читателей.

Когда использовать: данные часто читаются и редко изменяются. Читатели не мешают друг другу, писатель ждёт всех читателей.

import threading

class RWLock:
    def __init__(self):
        self._readers = 0
        self._read_lock = threading.Lock()      # защищает счётчик читателей
        self._write_lock = threading.Lock()     # эксклюзивный доступ писателя

    def acquire_read(self):
        with self._read_lock:
            self._readers += 1
            if self._readers == 1:
                self._write_lock.acquire()      # первый читатель блокирует писателя

    def release_read(self):
        with self._read_lock:
            self._readers -= 1
            if self._readers == 0:
                self._write_lock.release()      # последний читатель отпускает писателя

    def acquire_write(self):
        self._write_lock.acquire()

    def release_write(self):
        self._write_lock.release()

# Использование:
rwlock = RWLock()

def reader():
    rwlock.acquire_read()
    try:
        data = shared_data   # читаем (много потоков одновременно)
    finally:
        rwlock.release_read()

def writer():
    rwlock.acquire_write()
    try:
        shared_data = new_value  # пишем (только один поток)
    finally:
        rwlock.release_write()

Double-Checked Locking (ленивая инициализация)

Проверить без блокировки (быстро), если нужно — заблокировать и проверить ещё раз.

Когда использовать: ленивая инициализация дорогого ресурса, синглтон.

_resource = None
_lock = threading.Lock()

def get_resource():
    global _resource
    if _resource is None:            # быстрая проверка без блокировки
        with _lock:
            if _resource is None:    # повторная проверка под блокировкой
                _resource = create_expensive_resource()
    return _resource

Типовые проблемы многопоточности

1. Race Condition (состояние гонки)

Суть: результат зависит от порядка выполнения потоков. Два потока одновременно читают и пишут в одну переменную — кто «последний», тот и «победил». Результат непредсказуем.

# Гонка данных
counter = 0

def increment():
    global counter
    # Поток A: читает counter=0
    # Поток B: читает counter=0  (одновременно!)
    # Поток A: записывает counter=1
    # Поток B: записывает counter=1  (перезаписывает!)
    # Результат: 1 вместо 2
    counter += 1

# Решение: Lock
lock = threading.Lock()
def safe_increment():
    global counter
    with lock:
        counter += 1

# Пример гонки на check-then-act
if user not in cache:         # поток A проверяет — нет
    user = db.fetch(id)       # поток B проверяет — нет (A ещё не добавил)
    cache[user.id] = user     # поток A добавляет
                              # поток B тоже добавляет — дважды обратились к БД!

# Решение: атомарно через setdefault/get_or_create
user = cache.get(id)
if user is None:
    with lock:
        user = cache.get(id)          # повторная проверка (double check)
        if user is None:
            user = db.fetch(id)
            cache[id] = user

2. Deadlock (взаимная блокировка)

Суть: поток A ждёт ресурс, занятый потоком B; поток B ждёт ресурс, занятый потоком A. Оба ждут бесконечно.

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

def thread1():
    with lock_a:
        time.sleep(0.1)      # имитация работы
        with lock_b:         # ждёт lock_b, который у thread2
            ...

def thread2():
    with lock_b:
        time.sleep(0.1)
        with lock_a:         # ждёт lock_a, который у thread1 → DEADLOCK!
            ...

Решения:

1. Фиксированный порядок захвата: всегда захватывай блокировки в одном порядке
2. Таймаут: lock.acquire(timeout=5) — если не получилось за 5с, откатиться
3. try-lock: попробовать захватить второй lock; не получилось — отпустить первый и попробовать заново

def thread_fixed():
    with lock_a:                 # всегда сначала A
        with lock_b:             # потом B
            ...

# С таймаутом
def thread_timeout():
    while True:
        lock_a.acquire()
        if lock_b.acquire(timeout=1):
            try:
                ...              # работаем
                break
            finally:
                lock_b.release()
                lock_a.release()
        else:
            lock_a.release()     # откат
            time.sleep(0.01)     # ждём и пробуем снова

3. Livelock (живая блокировка)

Суть: потоки не заблокированы — они активно «уступают» друг другу, но никто не продвигается. Как два человека в коридоре: один шаг влево, другой шаг влево — снова друг перед другом.

# Упрощённый пример логики livelock
def thread_trylock():
    while True:
        if lock_a.acquire(blocking=False):
            if lock_b.acquire(blocking=False):
                break                  # получилось!
            lock_a.release()
        time.sleep(random.random())    # ждём случайное время
        # Если оба потока ждут одинаковое время → livelock

Решение: добавить случайную задержку (exponential backoff) или использовать фиксированный порядок захвата.

4. Starvation (голодание)

Суть: один или несколько потоков не могут получить доступ к ресурсу, потому что другие потоки постоянно его захватывают.

Причины:

• Приоритетное планирование (высокоприоритетные потоки не дают низкоприоритетным)
• Читатели не пускают писателя (read-write lock без защиты от starvation)
• notify() будит случайный поток — какой-то может ждать вечно

# Проблема: 100 коротких читателей могут не дать писателю войти
# Решение: очередь ожидания

import threading

class FairRWLock:
    def __init__(self):
        self._lock = threading.Lock()
        self._readers = 0
        self._write_ready = threading.Condition(self._lock)
        self._read_ready = threading.Condition(self._lock)
        self._writer_waiting = False

    def acquire_read(self):
        with self._lock:
            while self._writer_waiting:     # если ждёт писатель — читатели тоже ждут
                self._read_ready.wait()
            self._readers += 1

    def release_read(self):
        with self._lock:
            self._readers -= 1
            if self._readers == 0:
                self._write_ready.notify()  # последний читатель будит писателя

    def acquire_write(self):
        with self._lock:
            self._writer_waiting = True
            while self._readers > 0:
                self._write_ready.wait()

    def release_write(self):
        with self._lock:
            self._writer_waiting = False
            self._read_ready.notify_all()
            self._write_ready.notify()

5. Priority Inversion (инверсия приоритетов)

Суть: низкоприоритетный поток держит Lock, нужный высокоприоритетному. Среднеприоритетный поток вытесняет низкоприоритетный, не давая ему освободить Lock. Высокоприоритетный ждёт средне-приоритетный, который не имеет отношения к делу.

В Python (без real-time приоритетов) проявляется редко, но концептуально важно.

# Поток A (низкий приоритет): захватил Lock → делает долгую работу
# Поток B (средний приоритет): вытеснил A → крутится бесконечно
# Поток C (высокий приоритет): ждёт Lock, который держит A
# Итог: C ждёт B, хотя они не связаны!

# Решение: Priority Inheritance — временно повысить приоритет A до приоритета C
# (не реализовано в Python threading — используй multiprocessing с осторожностью)

6. False Sharing (ложное разделение)

Суть: две независимые переменные попали в одну строку кеша CPU. Поток A меняет переменную X → строка кеша инвалидируется → поток B, работающий с переменной Y в той же строке, теряет кеш. Производительность падает, хотя потоки не мешают друг другу логически.

Проявляется в multiprocessing.Value/Array и numpy в многопроцессорном режиме.

# Проблема: count1 и count2 в соседних ячейках памяти
counts = mp.Array('i', [0, 0, 0, 0])

def worker(counts, idx):
    for _ in range(10_000_000):
        counts[idx] += 1        # инвалидирует кеш для соседних индексов!

# Решение: разнести данные на расстояние > размера строки кеша (обычно 64 байта)
PADDING = 16  # int = 4 байта → 16 × 4 = 64 байта
padded = mp.Array('i', [0] * (4 * PADDING))

def worker_padded(padded, idx):
    for _ in range(10_000_000):
        padded[idx * PADDING] += 1   # каждый счётчик в своей строке кеша

7. ABA Problem

Суть: поток читает значение A, планирует его изменить, но другой поток меняет A → B → A обратно. Первый поток видит A и думает «ничего не изменилось», но на самом деле изменилось.

В Python встречается при использовании lock-free структур и queue.Queue.

# Упрощённый пример с check-then-act
if state == "READY":
    # ... другой поток: "READY" → "PROCESSING" → "READY"
    state = "DONE"              # мы думаем, что перешли из READY, но это уже другое READY!

Решение: использовать счётчик версий или compare-and-swap (CAS). В Python — через threading.Lock (защитить всю check-then-act операцию).

Общая стратегия предотвращения проблем

Частые ошибки

Ресурсы

• Раздел concurrency в документации Python — официальное описание модулей threading, multiprocessing, asyncio
• CPython на GitHub — исходный код Python (включая free-threading)

Что дальше

• Шпаргалка по Python: от основ до профи
• Шпаргалка по FastAPI: REST API, Pydantic, WebSocket
• Шпаргалка по Django: модели, views, ORM, админка

Связанные шпаргалки: Python, FastAPI, Django, SQL для начинающих