Многопоточность и GIL

Следующим шагом после профилирования, векторизации и JIT-компиляции является использование всех ядер процессора, а не одного. В Python это непросто, и рассмотрение необходимо начать с основ: что представляет собой процесс, что представляет собой поток и почему в обычной сборке CPython потоки не ускоряют вычисления.

Листинги главы выполнены в IPython на той же виртуальной машине с двумя ядрами под Python 3.12, что и в предыдущих главах; на ней же записаны демонстрации лекции «Оптимизация и параллельные вычисления». Сценарии раздела о сборке без GIL запущены на этой машине интерпретатором CPython 3.14 в свободнопоточной сборке. На машине с другим числом ядер числа будут иными.

Слайды к главе. Процессы и потоки, глобальная блокировка интерпретатора, пул процессов и сборка Python без GIL изложены также в седьмой части лекции «Оптимизация и параллельные вычисления» с демонстрациями в терминале; слайды лекции доступны на сайте книги и в PDF.

Необходимый минимум: процессы и потоки

Процесс

Процесс — запущенная программа, и операционная система выделяет каждому процессу собственное, изолированное от остальных состояние:

• виртуальное адресное пространство, то есть собственную память, скрытую от остальных процессов;
• системные ресурсы, например открытые файловые дескрипторы.

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

Поток

Поток исполняется независимо от других потоков, как и процесс, но существует внутри процесса и разделяет с ним адресное пространство и системные ресурсы. Собственными у каждого потока остаются указатель на исполняемую инструкцию, значения регистров и стек вызовов. Два потока одного процесса могут свободно работать с общими данными, поскольку они видят одни и те же объекты, размещённые в общей памяти.

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

Порядком исполнения как процессов, так и потоков распоряжается операционная система, поочерёдно выделяющая каждому кванты времени порядка миллисекунды, то есть миллионы тактов процессора.

Модуль threading

Поток в Python является обычным системным потоком, и его исполнением управляет операционная система, а не интерпретатор. Создать поток можно классом Thread, объявленным в модуле стандартной библиотеки threading.

import time
from threading import Thread

def countdown(n):
    for i in range(n):
        print(n - i - 1, "left")
        time.sleep(1)

t = Thread(target=countdown, args=(3,))

t.start()

2 left
1 left
0 left

Другой способ создать поток — наследование.

class CountdownThread(Thread):
    def __init__(self, n):
        super().__init__()
        self.n = n
        
    def run(self): # вызывается методом start
        for i in range(self.n):
            print(self.n - i - 1, "left")
            time.sleep(1)

t = CountdownThread(3)

t.start()

2 left
1 left
0 left

Этот подход ограничивает повторное использование кода: функциональность, заключённая в класс CountdownThread, доступна только в отдельном потоке.

У потока есть имя. По умолчанию оно имеет вид Thread-N, где N — порядковый номер, а начиная с Python 3.10 у потока с функцией в target к номеру добавляется имя этой функции. Имя, заданное разработчиком, лучше читается в журналах.

Thread().name

'Thread-3'

Thread(target=countdown).name

'Thread-4 (countdown)'

Thread(name="NumberCruncher").name

'NumberCruncher'

У каждого запущенного потока есть идентификатор ident — ненулевое число, уникальное среди работающих потоков; после завершения потока тот же идентификатор может получить другой поток. Номер, присвоенный потоку операционной системой, хранит атрибут native_id (с Python 3.8).

t = Thread()
t.start()
t.ident, t.native_id

(271580801397120, 39650)

Дождаться завершения потока позволяет метод join: вызывающий поток блокируется, пока t не завершит работу; повторный join на уже завершившемся потоке возвращается мгновенно.

t = Thread(target=time.sleep, args=(5, )) 
t.start()
t.join() # блокирует на 5 секунд
t.join() # выполняется мгновенно

Проверить, жив ли поток, можно методом is_alive.

t = Thread(target=time.sleep, args=(5, )) 
t.start()

t.is_alive() # False через 5 секунд

True

Поток, созданный с аргументом daemon=True, называется демоном: при выходе из интерпретатора он уничтожается автоматически, а не удерживает программу от завершения до окончания своей работы.

t = Thread(target=time.sleep, args=(5,), daemon=True)
t.start()

t.daemon

True

Встроенного способа принудительно завершить поток в Python нет, и это осознанное решение. Корректное завершение почти всегда означает освобождение ресурсов: поток мог открыть файл, дескриптор которого необходимо закрыть, или захватить примитив синхронизации, который необходимо освободить. Принудительное завершение оставило бы всё это в неопределённом состоянии, поэтому запрос на завершение передаётся потоку флагом, выставленным снаружи.

class Task:
    def __init__(self):
        self._running = True
    
    def terminate(self):
        self._running = False
    
    def run(self, n):
        while self._running:
            ...

Набор примитивов синхронизации в модуле threading стандартный:

• Lock, обычный мьютекс, обеспечивающий эксклюзивный доступ к разделяемому состоянию.
• RLock, рекурсивный мьютекс, разрешающий потоку, владеющему мьютексом, захватывать его более одного раза.
• Semaphore, вариация мьютекса, захватываемая не более фиксированного числа раз.
• BoundedSemaphore, семафор, следящий за тем, чтобы его захватывали и освобождали одинаковое число раз.
• Condition, условная переменная: поток ожидает в методе wait, пока другой поток не сообщит методом notify об изменении общего состояния.
• Event, событие: флаг, установки которого методом set потоки ожидают в методе wait.
• Barrier, барьер: потоки ожидают у него, пока барьера не достигнет заданное число потоков, и затем продолжают работу вместе.

Блокировки, семафоры и условные переменные реализуют единый интерфейс:

• метод acquire захватывает примитив синхронизации,
• а метод release освобождает его.

У событий и барьеров интерфейс собственный, поскольку они ничего не захватывают.

from threading import Lock


class SharedCounter:
    def __init__(self, value):
        self._value = value
        self._lock = Lock()
    
    def increment(self, delta=1):
        with self._lock:
            self._value += delta
    
    def get(self):
        return self._value

Захват и освобождение удобнее записывать блоком with: блокировка освобождается и тогда, когда внутри блока возникает исключение. При явных вызовах acquire и release исключение между ними, например при попытке прибавить к числу строку, оставило бы блокировку захваченной, и следующий вызов increment в любом потоке ожидал бы её бесконечно.

Модуль queue

Обмен данными между потоками с ручной расстановкой мьютексов неудобен. Обычно применяется готовая очередь; модуль queue реализует несколько потокобезопасных вариантов.

• Queue, очередь FIFO,
• LifoQueue, очередь LIFO, то есть стек,
• PriorityQueue, очередь с приоритетом: элементы, обычно пары (priority, item), выдаются не в порядке поступления, а по возрастанию приоритета.

Все методы, изменяющие состояние, работают под мьютексом. Queue хранит элементы в deque, а LifoQueue и PriorityQueue — в обычном списке, защищённом тем же мьютексом.

Классическая схема «производитель-потребитель» на очереди приведена ниже.

def worker(q):
    while True:
        item = q.get() # блокирующе ждёт следующий
        do_something(item) # элемент
        q.task_done() # уведомляет очередь о выполнении
        
def master(q):
    for item in source():
        q.put(item)

    # блокирующе ждёт, пока все элементы
    # очереди не будут обработаны
    q.join()

Модуль concurrent.futures

Модуль concurrent.futures предоставляет абстрактный класс Executor и его реализацию над пулом потоков, ThreadPoolExecutor. Интерфейс исполнителя состоит из трёх методов.

from concurrent.futures import *

executor = ThreadPoolExecutor(max_workers=4)

executor.submit(print, "Hello, world!")

Hello, world!
<Future at 0xf700558a6270 state=finished returned NoneType>

list(executor.map(print, ["Knock?", "Knock!"]))

Knock?
Knock!
[None, None]

executor.shutdown()

Исполнители поддерживают протокол менеджеров контекста: пул, открытый в with, закрывать вручную не требуется.

with ThreadPoolExecutor(max_workers=4) as executor:
    ...

Метод Executor.submit возвращает объект Future, обёртку над ещё не завершённым вычислением.

С Future можно выполнить следующие действия.

with ThreadPoolExecutor(max_workers=4) as executor:
    f = executor.submit(sorted, [4, 3, 1, 2])

• Запросить статус вычисления:

f.running(), f.done(), f.cancelled()

(False, True, False)

• Блокирующе дождаться результата вычисления:

print(f.result())

[1, 2, 3, 4]

print(f.exception())

None

• Добавить функцию, вызываемую после завершения вычисления:

f.add_done_callback(print)

<Future at 0xf7005422bb90 state=finished returned list>

Исключение, возникшее в задаче, вызывающий поток не прерывает: оно сохраняется в объекте Future, метод exception возвращает его, а result возбуждает повторно.

with ThreadPoolExecutor(max_workers=4) as executor:
    f = executor.submit(int, "x")
print(repr(f.exception()))

ValueError("invalid literal for int() with base 10: 'x'")

Поэтому результат каждой задачи необходимо запросить: иначе ошибка так и останется в объекте Future незамеченной.

Пример с модулем concurrent.futures: integrate

Проверим это на вычислительной задаче: численное интегрирование методом прямоугольников, циклом по заданному разбиению отрезка.

import math

def integrate(f, a, b, *, n_iter=1000):
    acc = 0
    step = (b - a) / n_iter
    for i in range(n_iter):
        acc += f(a + i * step) * step
    return acc

integrate(math.cos, 0.0, math.pi / 2)

1.0007851925466296

Нижний предел здесь и далее записан вещественным числом 0.0. При целом 0 выражение a + i * step складывает целое число с вещественным, а такое сложение специализирующий интерпретатор CPython 3.12 не ускоряет: на машине главы расчёт с n_iter=10**6 занимает 41 мс вместо 32 мс. Параллельные версии ниже получают вещественные границы частей, и целый предел поставил бы последовательную версию в невыгодное положение.

Параллельная версия делит отрезок на n_jobs частей и передаёт каждую пулу потоков; пул создаётся в блоке with и закрывается при выходе из функции.

from functools import partial

def integrate_async(f, a, b, *, n_jobs, n_iter=1000):
    with ThreadPoolExecutor(max_workers=n_jobs) as executor:
        spawn = partial(executor.submit, integrate, f,
                        n_iter=n_iter // n_jobs)
        step = (b - a) / n_jobs
        fs = [spawn(a + i * step, a + (i + 1) * step)
              for i in range(n_jobs)]
        return sum(f.result() for f in as_completed(fs))

integrate_async(math.cos, 0.0, math.pi / 2, n_jobs=2)

1.0007851925466305

Параллелизм и конкурентность

Различим два понятия. Конкурентностью называется способность программы вести несколько задач одновременно, переключаясь между ними, а параллелизмом — одновременное выполнение задач на нескольких ядрах. Конкурентная программа может выполняться и на одном ядре, параллельная — только на нескольких. Сравним производительность последовательной и параллельной версий integrate магической командой timeit.

%%timeit -n100
integrate(math.cos, 0.0, math.pi / 2, n_iter=10**6)

31.8 ms ± 213 μs per loop (mean ± std. dev. of 7 runs, 100 loops each)

%%timeit -n100
integrate_async(math.cos, 0.0, math.pi / 2, n_iter=10**6, n_jobs=2)

31.9 ms ± 243 μs per loop (mean ± std. dev. of 7 runs, 100 loops each)

Глобальная блокировка интерпретатора

Два потока вместо одного не дали никакого выигрыша: 31,9 мс против 31,8 мс у последовательной версии, хотя на двух ядрах можно было ожидать двукратного ускорения. Причиной является GIL (global interpreter lock, глобальная блокировка интерпретатора).

GIL представляет собой мьютекс, гарантирующий, что в каждый момент времени байт-код исполняет только один поток. Внутреннее состояние интерпретатора (в первую очередь счётчики ссылок у каждого объекта) не защищено от одновременного доступа, и глобальная блокировка является самым простым способом не позволить двум потокам одновременно нарушить его целостность. Поток, ожидающий блокировку, по истечении интервала переключения требует её освободить, и потоки исполняют байт-код по очереди; интервал по умолчанию составляет 5 мс.

import sys

sys.getswitchinterval()

0.005

Следствие: в обычной сборке CPython потоки не ускоряют вычисления на чистом Python. Сколько бы их ни было, байт-код исполняется по очереди, а к вычислениям добавляются накладные расходы на переключение, из-за которых многопоточная версия иногда оказывается даже медленнее однопоточной. Таким образом, потоки обычной сборки обеспечивают конкурентность, но не параллелизм.

Влияние GIL на разные задачи

Мешает ли GIL, зависит от того, чем занята программа.

Если программа вычисляет на чистом Python, потоки бесполезны. Обойти GIL можно несколькими путями, и все они рассматриваются ниже: перейти к отдельным процессам, у каждого из которых свой интерпретатор со своим GIL; перейти к скомпилированному коду, способному освобождать GIL; наконец, воспользоваться сборкой интерпретатора без GIL.

Если программа ожидает, то есть читает файл, получает данные по сети, опрашивает прибор, на время ожидания поток освобождает GIL, и остальные потоки работают. Для такой нагрузки потоки подходят, и проигрыша от блокировки нет.

NumPy на длинных операциях также освобождает GIL: пока перемножаются большие матрицы, интерпретатор свободен для остальных потоков. Поэтому многопоточность в сочетании с NumPy работает лучше, чем можно было бы ожидать.

Оговорка «в обычной сборке CPython» существенна: начиная с версии 3.13 интерпретатор собирается и вовсе без глобальной блокировки. Этой сборке посвящён раздел «Сборка Python без GIL» в конце главы, а до него всё изложенное относится к обычной сборке.

C и Cython: освобождение GIL

Скомпилированный код может освободить GIL явно, на то время, пока он не обращается к объектам Python. В Cython для этого предусмотрен блок with nogil, внутри которого допустимы только типы C, зато остальные потоки работают параллельно.

%load_ext Cython

%%cython

from libc.math cimport cos

def integrate(f, double a, double b, long n_iter):
    cdef double acc = 0
    cdef double step=(b-a)/n_iter
    cdef long i
    with nogil:
        for i in range(n_iter):
            acc += cos(a + i * step) * step
    return acc

Первый аргумент здесь оставлен только ради совместимости с integrate_async, а сама функция жёстко привязана к cos: вызвать функцию Python f внутри with nogil невозможно.

%%timeit -n100
integrate_async(math.cos, 0.0, math.pi / 2, n_iter=10**6, n_jobs=2)

1.79 ms ± 32.9 μs per loop (mean ± std. dev. of 7 runs, 100 loops each)

Те же два потока, тот же integrate_async, и 1,79 мс вместо 31,9. Этот выигрыш является составным. Потоков два, поэтому на снятие GIL приходится не более двукратного ускорения; остальное, примерно девятикратное, обеспечила компиляция в машинный код. Отсюда порядок действий: сначала применяется Cython, и только затем распараллеливается уже скомпилированный код.

О замерах ниже. Ячейка выше переопределила имя integrate: теперь под ним находится скомпилированная Cython-версия, жёстко привязанная к cos. Разделы о процессах и о joblib используют чистую Python-версию, которую следующий раздел записывает в модуль workers.py, поэтому для их повторения перезапускать блокнот не требуется.

Модуль multiprocessing

Процессы как способ обойти GIL

Поскольку глобальная блокировка одна на интерпретатор, самым прямым способом её обхода является запуск нескольких интерпретаторов. Именно это и обеспечивают процессы: у каждого свой GIL, друг другу они не мешают, и на восьми ядрах восемь процессов вычисляют одновременно.

Платой является изоляция. Процессы не разделяют память, поэтому всё, что необходимо передать между ними, приходится сериализовать и копировать. Для расчёта, в котором каждому процессу передаётся часть работы и забирается число, это несущественно; для задачи с большим общим состоянием накладные расходы на пересылку могут поглотить весь выигрыш.

За работу с процессами отвечает модуль multiprocessing, устроенный аналогично threading. Функция, которую выполняет дочерний процесс, передаётся ему по имени и должна импортироваться из модуля: при методах запуска spawn (macOS, Windows) и forkserver (Linux начиная с Python 3.14) функция из ячейки блокнота в дочернем процессе не находится, и он завершается ошибкой AttributeError. Поэтому функции примеров этого раздела записываются в файл workers.py и импортируются из него; туда же помещена чистая Python-версия integrate для следующих разделов.

%%writefile workers.py
import time


def countdown(n):
    for i in range(n):
        print(n - i - 1, "left")
        time.sleep(1)


def ponger(conn):
    conn.send("pong")


def integrate(f, a, b, *, n_iter=1000):
    acc = 0
    step = (b - a) / n_iter
    for i in range(n_iter):
        acc += f(a + i * step) * step
    return acc

Writing workers.py

import multiprocessing as mp
from workers import countdown, ponger, integrate

После импорта имя integrate снова указывает на чистую Python-версию.

p = mp.Process(target=countdown, args=(5, ))

p.start()

4 left
3 left
2 left
1 left
0 left

Примитивы синхронизации те же: мьютексы, семафоры, условные переменные. Для обмена данными между процессами предусмотрен Pipe, соединение двух процессов поверх сокетов.

parent_conn, child_conn = mp.Pipe()
p = mp.Process(target=ponger, args=(child_conn, ))

p.start()

parent_conn.recv()

'pong'

p.join()

Методы запуска процессов

Способ создания дочернего процесса задаёт метод запуска. Метод fork копирует родительский процесс целиком, вместе с его памятью и уже определёнными функциями. Метод spawn запускает новый интерпретатор, который импортирует главный модуль программы и получает функцию и аргументы в сериализованном виде. Метод forkserver при первом обращении запускает однопоточный процесс-сервер, а рабочие процессы порождает копированием этого сервера. По умолчанию в Windows и, начиная с Python 3.8, в macOS используется spawn, в Linux до версии 3.13 включительно — fork, а начиная с 3.14 — forkserver.

mp.get_start_method()

'fork'

Смена метода по умолчанию в Linux объясняется потоками. Копия процесса наследует блокировки, захваченные другими его потоками, но не сами эти потоки, и может навсегда остановиться на такой блокировке; начиная с Python 3.12 вызов fork в процессе с несколькими потоками сопровождается предупреждением DeprecationWarning. Метод выбирается функцией mp.set_start_method в начале программы или передаётся пулу явно: ProcessPoolExecutor(mp_context=mp.get_context("spawn")). При spawn и forkserver главный модуль программы импортируется в каждом рабочем процессе заново. Поэтому в сценарии создание процессов помещают под проверку if __name__ == "__main__":, иначе рабочий процесс, импортируя модуль, попытается снова создать дочерние процессы, и программа завершится ошибкой RuntimeError.

Оценим цену запуска. На машине главы, под Python 3.12 и 3.14, пул из двух процессов запускается методом fork за 1–4 мс, методом spawn — за 16–28 мс; forkserver тратит 19–28 мс на первый запуск, пока стартует сервер, а следующие пулы порождает за 3–6 мс.

Процессы и производительность

Реализация integrate_async на пуле потоков работала долго; рассмотрим пул процессов. ProcessPoolExecutor имеет тот же интерфейс, что и пул потоков, а функция integrate теперь импортирована из модуля workers.py.

from concurrent.futures import ProcessPoolExecutor

def integrate_async(executor, f, a, b, *, n_jobs, n_iter=1000):
    spawn = partial(executor.submit, integrate, f,
                    n_iter=n_iter // n_jobs)

    step = (b - a) / n_jobs
    fs=[spawn(a + i * step, a + (i + 1) * step)
        for i in range(n_jobs)]

    return sum(f.result() for f in as_completed(fs))

Пул принимается аргументом, а не создаётся внутри: пул, созданный внутри измеряемой функции, превратил бы замер вычислений в замер запуска процессов. Кроме того, ProcessPoolExecutor, оставленный без закрытия, оставляет процессы работающими, и за сотню итераций %%timeit их число на машине станет значительным. Поэтому пул открывается блоком with и прогревается первым вызовом, а последовательная и параллельная версии измеряются в одном прогоне.

with ProcessPoolExecutor(max_workers=2) as executor:
    integrate_async(executor, math.cos, 0.0, math.pi / 2,
                    n_iter=10**6, n_jobs=2)          # прогрев
    %timeit -n5 integrate(math.cos, 0.0, math.pi / 2, n_iter=10**6)
    %timeit -n5 integrate_async(executor, math.cos, 0.0, math.pi / 2, \
                                n_iter=10**6, n_jobs=2)

31.8 ms ± 88.3 μs per loop (mean ± std. dev. of 7 runs, 5 loops each)
19.9 ms ± 709 μs per loop (mean ± std. dev. of 7 runs, 5 loops each)

Последовательный расчёт занимает 31,8 мс, а два процесса — 19,9 мс, то есть в 1,6 раза меньше: у каждого процесса свой GIL, и оба ядра заняты вычислениями. До двукратного ускорения недостаёт накладных расходов: задачи и результаты пересылаются между процессами, и при расчёте длительностью в десятки миллисекунд эти расходы заметны.

Передача функции и аргументов

Функция и её аргументы передаются рабочему процессу сериализацией модулем pickle, а результат возвращается тем же путём. Функция при этом сериализуется по имени модуля и функции, поэтому в пуле выполняются функции верхнего уровня импортируемого модуля, а lambda и вложенная функция не сериализуются. Ошибка возникает не в момент вызова submit: сериализацию выполняет служебный поток пула, и исключение сохраняется в объекте Future до вызова result.

with ProcessPoolExecutor(max_workers=2) as executor:
    f = executor.submit(lambda x: x * x, 3)   # submit исключения не возбуждает
    try:
        f.result()                            # ошибка сериализации возбуждается здесь
    except Exception as exc:
        print(type(exc).__name__, exc)

PicklingError Can't pickle <function <lambda> at 0xf700564b99e0>: attribute lookup <lambda> on __main__ failed

Для lambda модуль pickle возбуждает PicklingError, а для вложенной функции до Python 3.13 включительно — AttributeError, начиная с 3.14 — тоже PicklingError. Без вызова result такая ошибка осталась бы незамеченной: блок with закрыл бы пул без сообщений.

Пул подынтерпретаторов

В Python 3.14 появился промежуточный вариант между пулом потоков и пулом процессов — InterpreterPoolExecutor из того же модуля concurrent.futures. Его рабочие потоки исполняют задачи в изолированных подынтерпретаторах одного процесса, и у каждого подынтерпретатора собственный GIL, поэтому вычисления на чистом Python выполняются параллельно и в обычной сборке. Общих изменяемых объектов у подынтерпретаторов нет: функция, аргументы и результат копируются из одного интерпретатора в другой. Функции без замыканий и без обращений к глобальным именам передаются непосредственно, остальные — через pickle, и такие функции, как и для пула процессов, должны импортироваться из модуля. Модуль расширения, не поддерживающий подынтерпретаторы, в них не импортируется: попытка завершается ошибкой ImportError.

Пакет joblib

Пакет joblib предоставляет параллельный аналог цикла for для независимых итераций, которые необходимо распределить по ядрам. Выбор между потоками и процессами задаётся аргументом backend конструктора: по умолчанию, начиная с версии 0.12, используется loky — собственный пул процессов joblib, сохраняемый между вызовами, а "threading" и "multiprocessing" указываются явно. Функция integrate в этом разделе — по-прежнему чистая Python-версия из workers.py.

from joblib import Parallel, delayed

def integrate_async(f, a, b, *, n_jobs, n_iter=1000, backend=None):
    step = (b - a) / n_jobs
    with Parallel(n_jobs=n_jobs, backend=backend) as parallel:
        fs = (delayed(integrate)(f, a + i * step,
                                 a + (i + 1) * step,
                                 n_iter=n_iter // n_jobs)
              for i in range(n_jobs))
        return sum(parallel(fs))   # пул существует только внутри одного вызова

%%timeit -n100
integrate_async(math.cos, 0.0, math.pi / 2, n_iter=10**6, n_jobs=2, backend="threading")

45 ms ± 426 μs per loop (mean ± std. dev. of 7 runs, 100 loops each)

%%timeit -n100
integrate_async(math.cos, 0.0, math.pi / 2, n_iter=10**6, n_jobs=2, backend="multiprocessing")

41.6 ms ± 405 μs per loop (mean ± std. dev. of 7 runs, 100 loops each)

Бэкенд по умолчанию запускает рабочие процессы при первом вызове, поэтому перед замером выполняется прогрев.

integrate_async(math.cos, 0.0, math.pi / 2, n_iter=10**6, n_jobs=2)   # прогрев: запуск процессов loky

1.0000007853979973

%%timeit -n100
integrate_async(math.cos, 0.0, math.pi / 2, n_iter=10**6, n_jobs=2)

23.4 ms ± 396 μs per loop (mean ± std. dev. of 7 runs, 100 loops each)

Сам по себе joblib ничего не ускоряет. На потоках работа упирается в тот же GIL, и к 31,8 мс последовательного расчёта добавляются расходы самого joblib: получается 45 мс. Бэкенд "multiprocessing" при каждом вызове integrate_async заново запускает и останавливает рабочие процессы: 41,6 мс против 19,9 мс у заранее запущенного пула из предыдущего раздела, то есть больше половины времени замера уходит на запуск и остановку. Бэкенд по умолчанию, loky, сохраняет процессы между вызовами и на двух процессах даёт ускорение — 23,4 мс против 31,8 мс; от ProcessPoolExecutor его отделяют накладные расходы joblib на каждый вызов.

JIT-компилятор numba

Numba из главы «Скорость выполнения программ» освобождает GIL только по запросу. Обычная функция @jit(nopython=True) без parallel=True удерживает GIL всё время вызова; чтобы несколько потоков Python выполняли её одновременно, её компилируют с параметром nogil=True. Покажем разницу на том же интеграле, записанном для Numba.

from numba import jit, prange

def cos_sum(a, b, n_iter):
    acc = 0.0
    step = (b - a) / n_iter
    for i in range(n_iter):
        acc += math.cos(a + i * step) * step
    return acc

held = jit(nopython=True)(cos_sum)                    # удерживает GIL
released = jit(nopython=True, nogil=True)(cos_sum)    # освобождает GIL

def in_two_threads(func, a, b, n_iter):
    mid = (a + b) / 2
    with ThreadPoolExecutor(max_workers=2) as executor:
        fs = [executor.submit(func, a, mid, n_iter // 2),
              executor.submit(func, mid, b, n_iter // 2)]
        return sum(f.result() for f in fs)

for func in (held, released):
    func(0.0, 1.0, 10)                                # компиляция для вещественных a и b

%timeit -n100 held(0.0, math.pi / 2, 10**6)
%timeit -n100 in_two_threads(held, 0.0, math.pi / 2, 10**6)
%timeit -n100 in_two_threads(released, 0.0, math.pi / 2, 10**6)

2.96 ms ± 20 μs per loop (mean ± std. dev. of 7 runs, 100 loops each)
3.1 ms ± 22.9 μs per loop (mean ± std. dev. of 7 runs, 100 loops each)
1.81 ms ± 60.4 μs per loop (mean ± std. dev. of 7 runs, 100 loops each)

Функция без nogil=True в двух потоках выполняется не быстрее, чем в одном: 3,1 мс против 2,96 мс, поскольку GIL удерживается всё время вызова и потоки работают по очереди. С nogil=True потоки вычисляют одновременно, и время сокращается до 1,81 мс.

Параметр parallel=True избавляет и от потоков Python: декоратор @jit(nopython=True, parallel=True) компилирует функцию в машинный код, а цикл prange распределяет по ядрам сама Numba, без участия GIL.

@jit(nopython=True, parallel=True, fastmath=True)
def integrate(a, b, n_iter=1000):
    acc = 0
    step = (b - a) / n_iter
    for i in prange(n_iter):
        acc += math.cos(a + i * step) * step
    return acc

integrate(0.0, math.pi / 2, n_iter=10**6)   # компиляция при первом вызове

1.0000007853979516

%%timeit -n100
integrate(0.0, math.pi / 2, n_iter=10**6)

1.83 ms ± 95.1 μs per loop (mean ± std. dev. of 7 runs, 100 loops each)

Цикл prange той же длины занимает 1,83 мс, столько же, сколько два потока с nogil=True: на двух ядрах параллелизм дал меньше двух раз, а основной выигрыш перед чистым Python с его 31,8 мс по-прежнему обеспечивает компиляция.

Сборка Python без GIL

Начиная с версии 3.13 интерпретатор CPython собирается и без глобальной блокировки (PEP 703). Такая сборка называется свободнопоточной (free-threaded), устанавливается рядом с обычной, и её исполняемый файл имеет суффикс t: python3.13t, python3.14t. В 3.13 сборка была экспериментальной, в 3.14 она поддерживается официально (PEP 779), но по умолчанию по-прежнему устанавливается обычная сборка. Потоки свободнопоточной сборки исполняют байт-код одновременно на всех ядрах, и вычисления на чистом Python распараллеливаются без процессов и без компиляции.

!python3.14t -VV

Python 3.14.8 free-threading build (main, Oct  1 2026, 21:01:11) [Clang 22.1.3 ]

Саму сборку определяют строка «free-threading build» в выводе python -VV и значение sysconfig.get_config_var("Py_GIL_DISABLED"), равное 1. Включена ли блокировка в работающем процессе, сообщает функция sys._is_gil_enabled() (с Python 3.13): в свободнопоточной сборке она возвращает False, если GIL не включён ключом -X gil=1, переменной окружения PYTHON_GIL=1 или импортом неподготовленного расширения.

!python3.14t -c "import sys, sysconfig; print(sysconfig.get_config_var('Py_GIL_DISABLED'), sys._is_gil_enabled())"
!PYTHON_GIL=1 python3.14t -c "import sys; print(sys._is_gil_enabled())"

1 False
True

Рассмотрим тот же интеграл на чистом Python в одном и в двух потоках. Сценарий ft.py печатает состояние блокировки и лучшее время из трёх запусков.

%%writefile ft.py
import math
import sys
import time
from concurrent.futures import ThreadPoolExecutor

from workers import integrate


def integrate_threads(f, a, b, *, n_jobs, n_iter):
    step = (b - a) / n_jobs
    with ThreadPoolExecutor(max_workers=n_jobs) as executor:
        fs = [executor.submit(integrate, f, a + i * step, a + (i + 1) * step,
                              n_iter=n_iter // n_jobs)
              for i in range(n_jobs)]
        return sum(f.result() for f in fs)


print("GIL включён:", sys._is_gil_enabled())
for n_jobs in (1, 2):
    times = []
    for _ in range(3):
        t0 = time.perf_counter()
        integrate_threads(math.cos, 0.0, math.pi / 2, n_jobs=n_jobs, n_iter=10**7)
        times.append(time.perf_counter() - t0)
    print(f"потоков: {n_jobs}, время {min(times):.2f} с")

Writing ft.py

!python3.14t -X gil=1 ft.py

GIL включён: True
потоков: 1, время 0.38 с
потоков: 2, время 0.38 с

!python3.14t ft.py

GIL включён: False
потоков: 1, время 0.38 с
потоков: 2, время 0.21 с

С включённой блокировкой два потока считают столько же, сколько один, — 0,38 с, а без неё — 0,21 с, в 1,8 раза быстрее: потоки исполняют байт-код одновременно на двух ядрах. Однопоточное время здесь нельзя сравнивать с замерами Python 3.12 выше: различаются и версия интерпретатора, и параметры его сборки, а цена отказа от GIL, о которой сказано ниже, относится к двум сборкам одной версии.

Отсутствие GIL не делает программу потокобезопасной. Встроенные типы — список, словарь, множество — защищены собственными внутренними блокировками, и их одновременное изменение не нарушает целостности объектов. Однако составное действие над общими данными, например counter += 1, по-прежнему состоит из чтения, сложения и записи, и эти шаги двух потоков могут перемежаться. Проверим это на счётчике, который два потока увеличивают по миллиону раз.

%%writefile race.py
import threading

counter = 0
lock = threading.Lock()


def bump(n, guarded):
    global counter
    for _ in range(n):
        if guarded:
            with lock:
                counter += 1
        else:
            counter += 1


for guarded in (False, True):
    counter = 0
    threads = [threading.Thread(target=bump, args=(10**6, guarded)) for _ in range(2)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    print("с Lock:" if guarded else "без Lock:", counter)

Writing race.py

!python3.14t race.py

без Lock: 1037029
с Lock: 2000000

Без Lock потеряно 962 971 увеличение из 2 000 000, почти половина: чтение, сложение и запись одного потока перемежаются с теми же шагами другого, и результат одного из них затирается. С Lock результат точный. Свободнопоточная сборка требует той же синхронизации доступа к общим данным, что и любая многопоточная программа.

Модули расширения на C должны объявить, что не полагаются на GIL: в C для этого служит слот Py_mod_gil при инициализации модуля, в Cython начиная с версии 3.1 — директива freethreading_compatible. При импорте модуля без такого объявления свободнопоточный интерпретатор снова включает блокировку для всего процесса и выводит предупреждение RuntimeWarning, в котором сказано, что оставить GIL выключенным на свой риск позволяют PYTHON_GIL=0 и -X gil=0. NumPy выпускает пакеты wheel для свободнопоточной сборки с версии 2.1, Numba — с версии 0.65, и обе библиотеки называют эту поддержку экспериментальной.

Платой за отказ от блокировки являются потери на однопоточном коде — по руководству по свободнопоточной сборке Python 3.14 в среднем от 1 % на macOS с процессором aarch64 до 8 % на Linux с x86-64 на наборе тестов pyperformance (в обзоре изменений 3.14 — «примерно 5–10 %», в 3.13 — около 40 %), — повышенный расход памяти и неготовность части C-расширений, написанных в расчёте на то, что блокировка защитит их сама. Таким образом, сборка без GIL позволяет распараллелить потоками код на чистом Python, но требует и потокобезопасной программы, и подготовленных расширений.

Резюме

Перед выбором инструмента необходимо ответить на один вопрос: программа вычисляет или ожидает?

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

Если программа вычисляет, потоки в обычной сборке CPython не помогут, сколько бы их ни было. Существует четыре пути: вынести расчёт в отдельные процессы через multiprocessing или ProcessPoolExecutor, где у каждого свой интерпретатор и свой GIL; перейти к скомпилированному коду на Cython, Numba или C, способному освобождать GIL; поручить расчёт NumPy, освобождающему блокировку самостоятельно, что обычно проще всего; наконец, начиная с Python 3.14 потоки на чистом Python распараллеливает официально поддерживаемая свободнопоточная сборка, если к ней готовы используемые расширения.

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