Лекция 6. Процессы и параллельные вычисления#
План:#
Что такое процесс: PID, раздельная память, запуск и завершение.
subprocess: вызвать внешнюю программу и получить её вывод.multiprocessing: запустить функцию, использовать пул, передать данные.Способы создания процессов:
spawn,fork,forkserver.Обмен между процессами:
Pipe,Queue,shared_memory.Другие способы использовать ядра: субинтерпретаторы и сборка без GIL.
В конце — раздел «Разбираем на семинаре»: Popen, multiprocessing.Pool,
устройство очередей и Manager.
Часть I. От программы к процессу#
Что такое процесс#
В Лекции 5 потоки выполняли разные функции, но работали с одними Python-объектами. Теперь хотим выполнять вычисления на разных ядрах даже при включённом GIL. Для этого можно запустить несколько процессов с отдельными интерпретаторами.
Программа — код на диске. Процесс — запущенный экземпляр программы:
у него есть собственное адресное пространство, открытые файлы и один или
несколько потоков выполнения. Один файл app.py можно запустить дважды —
получатся два процесса, каждый со своим состоянием.
У каждого процесса есть PID (process ID), идентификатор в операционной системе. Запустим из текущего Python ещё один Python и сравним PID:
import os
import subprocess
import sys
print("Родительский процесс:", os.getpid())
result = subprocess.run(
[sys.executable, "-c", "import os; print('Дочерний процесс:', os.getpid(), 'родитель:', os.getppid())"],
capture_output=True,
text=True,
check=True,
timeout=5,
)
print(result.stdout.strip())
print("Код завершения:", result.returncode)
Родительский процесс: 32649
Дочерний процесс: 32659 родитель: 32649
Код завершения: 0
PID различаются. Дочерний процесс знает PID родителя через os.getppid();
sys.executable запускает тот же Python, в котором выполняется ячейка,
а -c передаёт ему код строкой.
ОС может выполнять потоки этих процессов на разных ядрах. Само наличие
нескольких процессов ещё не гарантирует ускорения.
Адресное пространство — память, к которой процесс обращается по своим адресам. Обычный список в одном процессе недоступен другому просто по ссылке на него. У потоков одного процесса память общая; между процессами обмен нужно организовать явно.
subprocess.run: запуск, вывод, код завершения#
Нужно вызвать git, ffmpeg или другую готовую программу из Python.
subprocess.run() запускает её, ждёт завершения и возвращает CompletedProcess.
У программы есть стандартный ввод (stdin) и два стандартных потока вывода.
В stdout обычно пишут результат работы, а в stderr — диагностику:
ошибки, предупреждения, сообщения о ходе работы. Это отдельные потоки данных,
хотя при запуске из терминала оба обычно видны на одном экране.
В Python print() по умолчанию пишет в sys.stdout.
Для записи в другой поток передаём file=sys.stderr:
result = subprocess.run(
[
sys.executable,
"-c",
"import sys\n"
"print(42)\n"
"print('Обработка завершена', file=sys.stderr)\n",
],
capture_output=True,
text=True,
check=True,
timeout=5,
)
print("stdout:", repr(result.stdout))
print("stderr:", repr(result.stderr))
print("Код завершения:", result.returncode)
stdout: '42\n'
stderr: 'Обработка завершена\n'
Код завершения: 0
capture_output=True перехватывает оба потока раздельно: получаем
result.stdout и result.stderr. text=True декодирует байты в строки.
Разделение позволяет, например, передать результат следующей программе,
сохранив сообщения о ходе работы отдельно.
В примере stderr непустой, но процесс завершился успешно с кодом 0.
Запись в этот поток сама по себе не вызывает исключение в subprocess.
Код завершения сообщает, успешно ли закончилась программа: обычно 0 — успех,
ненулевое значение — ошибка. Это число задаёт сама программа; оно не связано
с возвращаемым значением Python-функции. check=True превращает ненулевой код
в исключение, чтобы вызывающий код не продолжал работу с неудачным результатом:
try:
subprocess.run(
[sys.executable, "-c", "import sys; print('Invalid input', file=sys.stderr); sys.exit(2)"],
capture_output=True,
text=True,
check=True,
timeout=5,
)
except subprocess.CalledProcessError as error:
print("Код завершения:", error.returncode)
print("Сообщение программы:", error.stderr.strip())
Код завершения: 2
Сообщение программы: Invalid input
Если внешняя программа зависла, не хочется ждать её бесконечно.
timeout ограничивает ожидание; при его истечении run() завершает дочерний
процесс, дожидается его и вызывает TimeoutExpired:
try:
subprocess.run(
[sys.executable, "-c", "import time; time.sleep(2)"],
check=True,
timeout=0.1,
)
except subprocess.TimeoutExpired:
print("Внешняя программа не уложилась в таймаут")
Внешняя программа не уложилась в таймаут
Таймаут не всегда прерывает сам этап создания процесса и не гарантирует остановку
всех потомков, которых программа успела запустить. Подробности — в
документации subprocess.run.
Аргументы и shell=True#
По умолчанию список передаёт программе отдельные аргументы. Оболочка (shell)
их не разбирает: ;, | и $(...) остаются частью строки.
С shell=True строку сначала выполняет оболочка по своим правилам.
Пусть пользователь ввёл имя файла, а мы хотим его вывести. Покажем инъекцию команд (shell injection) на безвредной дополнительной команде; пример рассчитан на POSIX:
if os.name == "posix":
user_input = "report.txt; printf INJECTED"
unsafe = subprocess.run(
f"echo {user_input}",
shell=True,
capture_output=True,
text=True,
check=True,
)
safe = subprocess.run(
["echo", user_input],
capture_output=True,
text=True,
check=True,
)
print("Через оболочку:", repr(unsafe.stdout))
print("Списком аргументов:", repr(safe.stdout))
else:
print("Демонстрация оболочки рассчитана на POSIX")
Через оболочку: 'report.txt\nINJECTED'
Списком аргументов: 'report.txt; printf INJECTED\n'
Оболочка увидела ; и выполнила две команды. При передаче списка echo получила
всё имя одним аргументом. Если нужны перенаправление вывода или соединение
программ каналом, их можно настроить через параметры subprocess.
Не подставляем пользовательский ввод в строку для shell=True.
Обычно достаточно списка аргументов. shlex.quote()
экранирует один аргумент для POSIX-оболочки, но не является универсальным решением для других оболочек.
Часть II. Python-функция в другом процессе#
Process: знакомые start() и join()#
Через subprocess можно запустить отдельный Python-скрипт, но аргументы функции
и результат пришлось бы передавать по собственному протоколу.
multiprocessing предоставляет готовые средства для запуска функций и обмена Python-объектами.
Начнём с одного процесса. Как у Thread, функция задаётся через target,
аргументы — через args, start() запускает работу, а join() ждёт завершения.
Выберем spawn: этот способ запуска доступен на Linux, macOS и Windows.
%%writefile /tmp/mp_intro.py
import multiprocessing as mp
import os
print(f"Импорт модуля: name={__name__}, pid={os.getpid()}")
def append_item(items: list[int]) -> int:
items.append(3)
print(f"Дочерний процесс: pid={os.getpid()}, родитель={os.getppid()}, список={items}")
return 42
if __name__ == "__main__":
ctx = mp.get_context("spawn")
items = [1, 2]
process = ctx.Process(target=append_item, args=(items,))
print("PID до запуска:", process.pid)
process.start()
print("PID после запуска:", process.pid)
waited = process.join()
print("Результат join():", waited)
print("Список у родителя:", items)
print("Код завершения:", process.exitcode)
Writing /tmp/mp_intro.py
!python /tmp/mp_intro.py
Импорт модуля: name=__main__, pid=32666
PID до запуска: None
PID после запуска: 32668
Импорт модуля: name=__mp_main__, pid=32668
Дочерний процесс: pid=32668, родитель=32666, список=[1, 2, 3]
Результат join(): None
Список у родителя: [1, 2]
Код завершения: 0
В дочернем процессе список стал [1, 2, 3], у родителя остался [1, 2]:
процесс получил отдельную копию. join() вернул None; значение 42
из target потерялось. exitcode == 0 означает нормальное завершение,
а не результат функции. Необработанное исключение в target даст код 1
и трейсбек в дочернем процессе, но не вызовет это исключение у родителя.
get_context("spawn") возвращает набор средств для выбранного способа запуска.
Из одного контекста будем создавать процессы, очереди и примитивы синхронизации.
Почему файл и if __name__ == "__main__"#
Строка «Импорт модуля» появилась дважды: свежий интерпретатор импортировал
главный модуль. Функции определились заново, но блок с запуском процесса
не выполнился, потому что имя модуля у дочернего процесса — __mp_main__.
Без проверки __name__ импорт снова попытался бы запустить процесс;
multiprocessing обнаруживает такой запуск во время инициализации и вызывает
RuntimeError. Проверка нужна и для forkserver.
Функция из ячейки ноутбука не находится в обычном импортируемом модуле.
Поэтому примеры с процессами сохраняем через %%writefile и запускаем как
скрипты. Для обычного subprocess.run() такой файл не требуется.
Зачем пул процессов#
При обработке сотен независимых входных данных создавать новый процесс
для каждой джобы дорого. Пул держит ограниченное число воркеров и переиспользует
их. ProcessPoolExecutor даёт знакомые по Лекции 5 map(), submit() и Future.
В обычной сборке CPython каждый процесс имеет свой интерпретатор и GIL: Python-код одного воркера не мешает другому выполнять Python-код на другом ядре. Сравним одинаковые вычисления последовательно и в пуле:
%%writefile /tmp/mp_procpool.py
from concurrent.futures import ProcessPoolExecutor
import multiprocessing as mp
import os
import time
def cpu_task(n: int) -> int:
total = 0
for i in range(n):
total += i * i
return total
if __name__ == "__main__":
inputs = [6_000_000] * 4
workers = min(4, os.process_cpu_count() or 1)
start = time.perf_counter()
sequential = [cpu_task(n) for n in inputs]
seq_time = time.perf_counter() - start
start = time.perf_counter()
with ProcessPoolExecutor(max_workers=workers, mp_context=mp.get_context("spawn")) as pool:
parallel = list(pool.map(cpu_task, inputs))
par_time = time.perf_counter() - start
print("Результаты совпали:", sequential == parallel)
print(f"Последовательно: {seq_time:.2f} с; процессы ({workers}): {par_time:.2f} с")
print(f"Ускорение: {seq_time / par_time:.2f}x")
Writing /tmp/mp_procpool.py
!python /tmp/mp_procpool.py
Результаты совпали: True
Последовательно: 0.67 с; процессы (4): 0.25 с
Ускорение: 2.66x
В Лекции 5 при включённом GIL потоки не ускоряли такой цикл. Здесь вычисления могут идти параллельно, но в измерение также входят запуск воркеров, импорты, передача аргументов и результатов. На одном доступном ядре выигрыша не будет; даже на нескольких ускорение не обязано быть линейным.
map() выдаёт результаты в порядке входных данных. Если нужна обработка по мере
готовности, используем submit() и as_completed(), как с пулом потоков.
Исключение из джобы возникает при получении результата через Future.result()
или при обходе результатов map(). Выход из with ждёт завершения отправленных джоб.
Больше воркеров не всегда быстрее. Есть доступные ядра, лимиты памяти и цена обмена данными. Универсального порога вроде «джоба должна длиться не меньше 100 мс» нет: измеряем свою нагрузку, переиспользуем пул, мелкие джобы объединяем в пакеты.
Что именно передаётся: pickle#
Пул должен перенести функцию и её аргументы в другой процесс, затем вернуть
результат. Для этого он использует сериализацию: представляет данные
последовательностью байтов. В стандартном пуле это делает pickle.
import pickle
original = {"values": [1, 2]}
wire = pickle.dumps(original)
restored = pickle.loads(wire)
restored["values"].append(3)
print("Байтов в сообщении:", len(wire))
print("Исходный объект:", original)
print("Восстановленный объект:", restored)
Байтов в сообщении: 32
Исходный объект: {'values': [1, 2]}
Восстановленный объект: {'values': [1, 2, 3]}
Сериализация восстанавливает данные, а не общую ссылку. Для функции pickle
сохраняет имя модуля и имя функции, поэтому получатель должен суметь её импортировать.
Покажем, как ошибка передачи появляется через Future:
%%writefile /tmp/mp_functions.py
from concurrent.futures import ProcessPoolExecutor
import multiprocessing as mp
import pickle
def double(x: int) -> int:
return x * 2
if __name__ == "__main__":
print("Сериализованная функция:", pickle.dumps(double))
with ProcessPoolExecutor(max_workers=1, mp_context=mp.get_context("spawn")) as pool:
future = pool.submit(lambda x: x * 2, 21)
try:
print(future.result())
except pickle.PicklingError as error:
print("Ошибка передачи функции:", error)
print("Функция уровня модуля:", pool.submit(double, 21).result())
Writing /tmp/mp_functions.py
!python /tmp/mp_functions.py
Сериализованная функция: b'\x80\x05\x95\x17\x00\x00\x00\x00\x00\x00\x00\x8c\x08__main__\x94\x8c\x06double\x94\x93\x94.'
Ошибка передачи функции: Can't pickle <function <lambda> at 0x106181220>: it's not found as __main__.<lambda>
Функция уровня модуля: 42
У lambda нет подходящего имени для восстановления. Вложенные функции и локальные
классы тоже обычно не подходят для стандартного pickle. Выносим функцию на уровень
модуля и явно передаём аргументы. Аргументы и возвращаемое значение также должны
поддерживать сериализацию; например, открытый файл передать как обычный аргумент нельзя.
Даже при fork пул пересылает джобы через очередь, поэтому это ограничение сохраняется.
Часть III. Откуда берётся дочерний процесс#
spawn: новый интерпретатор#
В примерах уже использовали spawn: запускается новый интерпретатор,
импортирует главный модуль и получает функцию с аргументами.
Глобальные переменные заново инициализируются при импорте; изменения,
которые родитель сделал после импорта, автоматически к дочернему процессу не попадут.
Новый процесс всё же получает необходимое окружение, например переменные окружения и текущий каталог. «Свежий интерпретатор» не означает «ничего не унаследовано». Цена запуска зависит в том числе от импортов программы.
fork: снимок состояния родителя#
На POSIX можно вызвать fork(): ОС создаёт процесс со снимком адресного
пространства родителя. В дочернем процессе продолжается только поток,
который вызвал fork; остальные потоки туда не переходят.
Память обычно копируется по принципу копирования при записи (copy-on-write): родитель и потомок сначала используют одни физические страницы, а при записи получают отдельные копии изменяемых страниц. Это экономит копирование при запуске, но не делает изменения Python-объектов общими.
Родитель: список [1, 2] Потомок: список [1, 2]
| |
+-------- общие страницы -------+
|
запись append(3)
|
отдельная копия страницы
В снимок могут попасть и захваченные блокировки, и открытые соединения. Если блокировку держал другой поток, в дочернем процессе уже некому её отпустить. Унаследованное сетевое соединение тоже не превращается в два независимых соединения.
forkserver: снимок отдельного сервера#
Хотим сохранить преимущества fork, но не снимать состояние многопоточного
приложения. Для этого запускается отдельный сервер со свежим интерпретатором;
по запросу он создаёт воркеров через fork() из своего состояния.
Родитель -- запрос --> сервер forkserver
|
+-- fork() --> воркер 1
+-- fork() --> воркер 2
Сервер обычно однопоточный. Он может предварительно импортировать модули;
если такой импорт создаёт потоки, предпосылка безопасности нарушается.
Воркер получает функцию и аргументы от родителя, но не произвольные изменения
его глобальных переменных. Проверка __name__ по-прежнему нужна.
Что выберет Python 3.14#
import multiprocessing as mp
print("Способ по умолчанию:", mp.get_start_method())
print("Доступные способы:", mp.get_all_start_methods())
Способ по умолчанию: spawn
Доступные способы: ['spawn', 'fork', 'forkserver']
На Windows и macOS по умолчанию используется spawn; на Linux и других
поддерживаемых POSIX-платформах — forkserver. В Python 3.14 fork больше
не является способом по умолчанию ни на одной платформе. forkserver
доступен и на некоторых POSIX-системах помимо Linux; смотрим результат проверки.
Документация способов запуска.
Живой дедлок: fork после запуска потока#
Поток родителя захватит threading.Lock. Сначала дождёмся этого через Event,
чтобы результат не зависел от случайной задержки, затем создадим процесс.
В дочернем процессе попробуем захватить ту же блокировку:
%%writefile /tmp/fork_deadlock.py
import multiprocessing as mp
import sys
import threading
lock = threading.Lock()
locked = threading.Event()
release = threading.Event()
def hold_lock() -> None:
with lock:
locked.set()
release.wait()
def child_job() -> None:
print("Дочерний процесс: пытаюсь взять блокировку", flush=True)
with lock:
print("Дочерний процесс: блокировка захвачена", flush=True)
if __name__ == "__main__":
method = sys.argv[1]
if method not in mp.get_all_start_methods():
print(f"Способ {method} недоступен на этой платформе")
else:
thread = threading.Thread(target=hold_lock)
thread.start()
locked.wait()
process = mp.get_context(method).Process(target=child_job)
process.start()
process.join(timeout=2)
if process.is_alive():
print("Родитель: ожидание истекло, дочерний процесс заблокирован")
process.terminate()
process.join()
print("Код завершения:", process.exitcode)
release.set()
thread.join()
Writing /tmp/fork_deadlock.py
!python /tmp/fork_deadlock.py fork
Дочерний процесс: пытаюсь взять блокировку
Родитель: ожидание истекло, дочерний процесс заблокирован
Код завершения: -15
У дочернего процесса есть захваченный lock, но нет потока hold_lock,
который мог бы его отпустить. Даже освобождение блокировки у родителя
не исправило бы её копию у потомка. join(timeout=2) ограничил ожидание
родителя, а не работу дочернего процесса: для остановки понадобился отдельный
terminate() и затем join().
Меняем только способ запуска:
!python /tmp/fork_deadlock.py forkserver
Дочерний процесс: пытаюсь взять блокировку
Дочерний процесс: блокировка захвачена
Код завершения: 0
Теперь воркер создаётся из состояния сервера. Захваченная блокировка родителя
туда не попала; блокировка из импортированного модуля свободна. spawn в этом
примере тоже сработает. Это не передача threading.Lock между процессами:
для общей блокировки нужен отдельный примитив из multiprocessing.
Принудительное завершение может оставить очередь или общую блокировку повреждённой,
а код очистки дочернего процесса не обязан выполниться. В рабочих программах сначала
организуем нормальную остановку: сообщение через очередь или общий сигнал через
ctx.Event().
Для переносимых примеров выбираем spawn, для приложения можно использовать
контекст по умолчанию. В библиотеке даём вызывающему коду передать контекст;
глобальный set_start_method() может конфликтовать с другими библиотеками.
Процесс и его общие примитивы создаём из одного контекста.
Часть IV. Как получить результат и обмениваться данными#
IPC (inter-process communication) — обмен между процессами.
У Process нет аналога Future.result(), поэтому способ возврата результата
выбираем сами. Для отдельных джоб это уже делает пул; для длительно работающих
процессов пригодятся каналы и очереди.
Pipe: два конца канала#
Одному воркеру отправим число и получим квадрат. Pipe() возвращает два конца
двунаправленного канала: send() передаёт объект, recv() ждёт сообщение.
%%writefile /tmp/mp_pipe.py
import multiprocessing as mp
from multiprocessing.connection import Connection
def worker(conn: Connection) -> None:
value = conn.recv()
conn.send(value * value)
conn.close()
if __name__ == "__main__":
ctx = mp.get_context("spawn")
parent_conn, child_conn = ctx.Pipe()
process = ctx.Process(target=worker, args=(child_conn,))
process.start()
child_conn.close()
parent_conn.send(21)
print("Результат:", parent_conn.recv())
parent_conn.close()
process.join()
print("Код завершения:", process.exitcode)
Writing /tmp/mp_pipe.py
!python /tmp/mp_pipe.py
Результат: 441
Код завершения: 0
Закрываем концы, которыми больше не пользуемся. send() сериализует объект
через pickle, recv() восстанавливает его. Разные концы можно использовать
одновременно; несколько одновременных чтений или несколько записей в один
конец без внешней синхронизации могут повредить сообщения.
Queue: раздать джобы нескольким воркерам#
С каналом пришлось бы самому решать, какому воркеру отправить следующую джобу.
Очередь позволяет нескольким процессам безопасно извлекать сообщения:
каждую джобу получает один из воркеров. Интерфейс знаком по queue.Queue,
но объекты передаются между процессами, а не по общей ссылке.
%%writefile /tmp/mp_queue.py
import multiprocessing as mp
import os
from multiprocessing.queues import Queue
def consumer(q: Queue[int | None]) -> None:
while True:
item = q.get()
if item is None:
break
print(f"pid={os.getpid()}: джоба {item}, результат {item * item}", flush=True)
if __name__ == "__main__":
ctx = mp.get_context("spawn")
q = ctx.Queue(maxsize=2)
processes = [ctx.Process(target=consumer, args=(q,)) for _ in range(2)]
for process in processes:
process.start()
for item in range(6):
q.put(item)
for _ in processes:
q.put(None)
for process in processes:
process.join()
print("Коды завершения:", [process.exitcode for process in processes])
q.close()
q.join_thread()
Writing /tmp/mp_queue.py
!python /tmp/mp_queue.py
pid=32716: джоба 0, результат 0
pid=32715: джоба 1, результат 1
pid=32716: джоба 2, результат 4
pid=32715: джоба 3, результат 9
pid=32716: джоба 4, результат 16
pid=32715: джоба 5, результат 25
Коды завершения: [0, 0]
None — маркер завершения (sentinel): одному воркеру он говорит выйти из цикла.
Воркеров два, поэтому маркеров тоже два. Распределение джоб и порядок печати
могут меняться; равную долю работы каждому воркеру очередь не обещает.
maxsize=2 ограничивает число сообщений в очереди: если воркеры не успевают,
put() ждёт свободного места. Размер самих сообщений при этом не ограничен.
Для завершения используем явный протокол, а не проверку empty():
состояние очереди может измениться сразу после проверки.
Часть V. Параллелизм без нескольких процессов#
Субинтерпретаторы: один PID, отдельные состояния Python#
Процессы дали независимые интерпретаторы, но пришлось платить за создание процессов и передачу данных. Можно ли получить несколько интерпретаторов внутри одного процесса?
Субинтерпретатор (subinterpreter) имеет свои модули, глобальные переменные
и состояние Python. PEP 684 добавил возможность
отдельного GIL для интерпретатора в CPython 3.12; в Python 3.14
PEP 734 добавил публичный модуль
concurrent.interpreters и InterpreterPoolExecutor.
Сравним PID и изменяемые данные. Через prepare_main() передадим сериализованный
список как bytes, а очередь для ответа создадим через interpreters.create_queue():
from concurrent import interpreters
import os
import pickle
items = [1, 2]
q = interpreters.create_queue()
interp = interpreters.create()
try:
interp.prepare_main(data=pickle.dumps(items), q=q)
interp.exec(
"import os, pickle\n"
"items = pickle.loads(data)\n"
"items.append(3)\n"
"q.put((os.getpid(), tuple(items)))\n"
)
other_pid, other_items = q.get()
print("Главный интерпретатор:", os.getpid(), items)
print("Другой интерпретатор:", other_pid, list(other_items))
finally:
interp.close()
Главный интерпретатор: 32649 [1, 2]
Другой интерпретатор: 32649 [1, 2, 3]
PID один, но изменение списка не затронуло главный интерпретатор.
Обычные изменяемые Python-объекты не становятся общими автоматически.
В примере явно передали данные через pickle, а ответ — кортеж из поддерживаемых типов.
В то же время адресное пространство процесса общее: это изоляция состояния
Python, а не граница защиты, которую даёт отдельный процесс ОС.
interp.exec() выполняет код в текущем потоке и ждёт завершения.
Само создание интерпретаторов не создаёт параллелизм: для него нужно выполнять
их в разных потоках. Готовый вариант — InterpreterPoolExecutor:
%%writefile /tmp/mp_compare.py
from concurrent.futures import InterpreterPoolExecutor, ThreadPoolExecutor
import os
import time
from mp_procpool import cpu_task
if __name__ == "__main__":
inputs = [6_000_000] * 4
workers = min(4, os.process_cpu_count() or 1)
expected = [cpu_task(n) for n in inputs]
for executor in (ThreadPoolExecutor, InterpreterPoolExecutor):
start = time.perf_counter()
with executor(max_workers=workers) as pool:
results = list(pool.map(cpu_task, inputs))
elapsed = time.perf_counter() - start
print(f"{executor.__name__}: {elapsed:.2f} с, результаты совпали: {results == expected}")
Writing /tmp/mp_compare.py
!python /tmp/mp_compare.py
ThreadPoolExecutor: 0.62 с, результаты совпали: True
InterpreterPoolExecutor: 0.20 с, результаты совпали: True
Каждый воркер этого пула — поток со своим интерпретатором и отдельным GIL,
поэтому вычисления могут использовать разные ядра. Функции, аргументы и
результаты InterpreterPoolExecutor тоже передаются через pickle;
это не бесплатная замена пулу процессов.
Не все расширения на C поддерживают несколько интерпретаторов: проверяем
совместимость нужных библиотек. Авария расширения может завершить весь процесс,
включая главный интерпретатор.
Документация InterpreterPoolExecutor.
Сборка без GIL: общие объекты и параллельные потоки#
Субинтерпретаторы разнесли состояние Python. Другой подход — разрешить потокам
одного интерпретатора выполнять Python-код одновременно, сохранив общие объекты.
Для этого PEP 703 добавил сборку
free-threaded; в Python 3.14 она получила поддерживаемый статус по
PEP 779, но остаётся отдельной сборкой.
Как в Лекции 5, проверим возможности сборки и текущий режим отдельно:
import sys
import sysconfig
print("Сборка поддерживает отключение GIL:", bool(sysconfig.get_config_var("Py_GIL_DISABLED")))
print("GIL сейчас включён:", sys._is_gil_enabled())
Сборка поддерживает отключение GIL: False
GIL сейчас включён: True
Для самостоятельного сравнения можно установить сборку через uv
и запустить тот же файл с вычислениями:
uv python install 3.14t
uv run --no-project --isolated --python 3.14t python /tmp/mp_compare.py
Без GIL обычный ThreadPoolExecutor тоже может ускорить CPU-bound Python-код,
а объекты остаются общими. Синхронизация из Лекции 5 по-прежнему нужна:
внутренние блокировки контейнеров не делают составную операцию вроде
«проверить, затем изменить» атомарной.
Поддержку проверяем на своих зависимостях: расширение на C без поддержки этого режима может включить GIL при импорте. Ускорение зависит от нагрузки, а новая организация памяти тоже имеет накладные расходы. Документация сборки без GIL.
Что выбрать для вычислений#
Сравниваем CPU-bound код на Python. Если расширение на C освобождает GIL, потоки могут использовать разные ядра и в обычной сборке.
Инструмент |
Python-код на разных ядрах |
Изменяемые Python-объекты |
Граница отдельного процесса ОС |
|---|---|---|---|
|
нет, внутри одного интерпретатора |
общие |
нет |
|
да |
отдельные копии, явный обмен |
да |
|
да |
отдельные состояния интерпретаторов, явный обмен |
нет |
|
да |
общие, нужна синхронизация |
нет |
Для обычной сборки и независимых вычислений начинаем с ProcessPoolExecutor.
Если важны изоляция от аварий расширений и отдельные ресурсы ОС, процессы полезны
даже при доступной сборке без GIL. Для субинтерпретаторов и сборки без GIL сначала
проверяем совместимость зависимостей и измеряем выигрыш.
Для внешней программы используем subprocess; её тоже можно запускать параллельно
с другими. Для ожидания ввода-вывода пригодятся потоки и асинхронное выполнение —
его разберём дальше в курсе.
Разбираем на семинаре#
Popen: читать вывод до завершения программы#
run() подходит, когда нужен итоговый результат. Если программа долго работает
и печатает ход обработки, хочется видеть строки сразу. Popen запускает процесс
и возвращает управление; чтение вывода и ожидание организуем сами:
with subprocess.Popen(
[sys.executable, "-u", "-c", "import time\nfor i in range(3):\n print(i)\n time.sleep(0.1)"],
stdout=subprocess.PIPE,
text=True,
) as process:
for line in process.stdout:
print("Получили строку:", line.strip())
process.wait()
print("Код завершения:", process.returncode)
Получили строку: 0
Получили строку: 1
Получили строку: 2
Код завершения: 0
-u отключает буферизацию вывода дочернего Python: иначе строки могли бы
дойти только при завершении. При выходе из with закрываются потоки
и ожидается процесс. В отличие от run(), Popen.wait(timeout=...)
при истечении таймаута сам процесс не завершает.
Если отдельно читать stdout, а stderr тоже направить в PIPE, дочерний
процесс может заблокироваться при заполнении непрочитанного канала ошибок.
Когда оба вывода нужны целиком, используем communicate(); для единого потока
можно направить stderr=subprocess.STDOUT.
multiprocessing.Pool: другой интерфейс пула#
Этот интерфейс встречается в существующем коде. map() собирает результаты
по порядку, а imap_unordered() выдаёт их по мере готовности:
%%writefile /tmp/mp_pool.py
import multiprocessing as mp
import time
def square(x: int) -> int:
time.sleep(0.05 * (3 - x % 3))
return x * x
if __name__ == "__main__":
with mp.get_context("spawn").Pool(processes=2) as pool:
print("map:", pool.map(square, range(6)))
for result in pool.imap_unordered(square, range(6)):
print("Готов результат:", result)
Writing /tmp/mp_pool.py
!python /tmp/mp_pool.py
map: [0, 1, 4, 9, 16, 25]
Готов результат: 1
Готов результат: 4
Готов результат: 0
Готов результат: 16
Готов результат: 9
Готов результат: 25
У Pool и ProcessPoolExecutor различается завершение: выход из with Pool(...)
вызывает terminate(), поэтому незавершённая работа может быть оборвана.
В этом примере результаты получены до выхода. У ProcessPoolExecutor выход из
контекста по умолчанию ждёт отправленные джобы.
Пакеты мелких джоб#
У каждого сообщения есть цена сериализации и передачи. В ProcessPoolExecutor.map
параметр chunksize объединяет несколько входных элементов в одну передачу воркеру:
results = pool.map(cpu_task, inputs, chunksize=16)
Больший пакет уменьшает число сообщений, но ухудшает распределение, если джобы
сильно различаются по длительности. Значение выбираем по измерениям;
для ThreadPoolExecutor и InterpreterPoolExecutor этот параметр не влияет на работу.
Что внутри multiprocessing.Queue#
put() сначала ждёт свободное место по счётчику, затем добавляет объект
в локальный буфер. Фоновый поток отправки (feeder thread) сериализует объекты
и пишет их в канал. get() читает из канала под блокировкой и восстанавливает объект.
Фоновый поток работает на стороне отправителя, а не получателя.
Из этого следует неочевидная ловушка: put() уже вернулся, но отправитель
не может завершиться, пока фоновый поток не отправит буфер. Если родитель
ждёт отправителя, а канал заполнен, оба останутся ждать.
Покажем ошибочный порядок, затем исправим его, прочитав сообщение:
%%writefile /tmp/mp_queue_join.py
import multiprocessing as mp
from multiprocessing.queues import Queue
def send_large(q: Queue[bytes]) -> None:
q.put(b"x" * 1_000_000)
if __name__ == "__main__":
ctx = mp.get_context("spawn")
q = ctx.Queue()
process = ctx.Process(target=send_large, args=(q,))
process.start()
process.join(timeout=1)
print("После join до чтения процесс ещё работает:", process.is_alive())
print("Прочитали байтов:", len(q.get()))
process.join()
print("После чтения код завершения:", process.exitcode)
q.close()
q.join_thread()
Writing /tmp/mp_queue_join.py
!python /tmp/mp_queue_join.py
После join до чтения процесс ещё работает: True
Прочитали байтов: 1000000
После чтения код завершения: 0
Сначала освобождаем канал чтением, затем ждём отправителя. Описание этой ловушки.
SimpleQueue не использует фоновый поток: put() сам сериализует и отправляет
объект, поэтому может блокироваться на записи. У него нет ограничения maxsize
и таймаутов put()/get(), но есть empty().
JoinableQueue добавляет task_done() и join() для ожидания обработки
сообщений, как queue.Queue из Лекции 5. Это другое ожидание, чем Process.join():
воркер может обработать все джобы и остаться работать.
Manager: общий словарь через отдельный процесс#
Иногда нескольким процессам нужен небольшой общий словарь, а проектировать
формат байтового буфера неудобно. Manager хранит Python-объекты в отдельном
процессе и выдаёт прокси: каждый вызов обращается к тому процессу.
%%writefile /tmp/mp_manager.py
import multiprocessing as mp
from multiprocessing.managers import DictProxy
def worker(results: DictProxy[str, int], name: str, value: int) -> None:
results[name] = value * value
if __name__ == "__main__":
ctx = mp.get_context("spawn")
with ctx.Manager() as manager:
results = manager.dict()
processes = [
ctx.Process(target=worker, args=(results, f"worker-{i}", i))
for i in range(3)
]
for process in processes:
process.start()
for process in processes:
process.join()
print("Общий словарь:", dict(results))
Writing /tmp/mp_manager.py
!python /tmp/mp_manager.py
Общий словарь: {'worker-2': 4, 'worker-0': 0, 'worker-1': 1}
Это не общая ссылка на обычный dict: объект живёт в процессе менеджера,
а аргументы и ответы передаются сообщениями. Удобно для небольших данных
координации, но часто обращаться так к большим данным дорого.
Последовательность операций вроде results["count"] += 1 не становится атомарной:
чтение и запись — отдельные обращения, при конкуренции нужна блокировка.
manager.Queue() тоже возвращает прокси. Её, например, можно передавать
как аргумент джобы уже работающего пула. Обычную ctx.Queue() передаём воркерам
при создании процессов, а не произвольной джобой через submit().