--- jupytext: formats: md:myst text_representation: extension: .md format_name: myst kernelspec: display_name: Python 3.14 language: python name: python3 --- ```{code-cell} python :tags: [remove-cell] import os import sys os.environ["PATH"] = f"{os.path.dirname(sys.executable)}:{os.environ['PATH']}" os.environ.pop("VIRTUAL_ENV", None) ``` # Лекция 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: ```{code-cell} python 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) ``` 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`: ```{code-cell} python 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) ``` `capture_output=True` перехватывает оба потока раздельно: получаем `result.stdout` и `result.stderr`. `text=True` декодирует байты в строки. Разделение позволяет, например, передать результат следующей программе, сохранив сообщения о ходе работы отдельно. В примере `stderr` непустой, но процесс завершился успешно с кодом `0`. Запись в этот поток сама по себе не вызывает исключение в `subprocess`. **Код завершения** сообщает, успешно ли закончилась программа: обычно `0` — успех, ненулевое значение — ошибка. Это число задаёт сама программа; оно не связано с возвращаемым значением Python-функции. `check=True` превращает ненулевой код в исключение, чтобы вызывающий код не продолжал работу с неудачным результатом: ```{code-cell} python 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()) ``` Если внешняя программа зависла, не хочется ждать её бесконечно. `timeout` ограничивает ожидание; при его истечении `run()` завершает дочерний процесс, дожидается его и вызывает `TimeoutExpired`: ```{code-cell} python try: subprocess.run( [sys.executable, "-c", "import time; time.sleep(2)"], check=True, timeout=0.1, ) except subprocess.TimeoutExpired: print("Внешняя программа не уложилась в таймаут") ``` Таймаут не всегда прерывает сам этап создания процесса и не гарантирует остановку всех потомков, которых программа успела запустить. Подробности — в [документации `subprocess.run`](https://docs.python.org/3.14/library/subprocess.html#subprocess.run). ### Аргументы и `shell=True` По умолчанию список передаёт программе отдельные аргументы. Оболочка (shell) их не разбирает: `;`, `|` и `$(...)` остаются частью строки. С `shell=True` строку сначала выполняет оболочка по своим правилам. Пусть пользователь ввёл имя файла, а мы хотим его вывести. Покажем инъекцию команд (shell injection) на безвредной дополнительной команде; пример рассчитан на POSIX: ```{code-cell} python 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") ``` Оболочка увидела `;` и выполнила две команды. При передаче списка `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. ```{code-cell} python %%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) ``` ```{code-cell} python !python /tmp/mp_intro.py ``` В дочернем процессе список стал `[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-код на другом ядре. Сравним одинаковые вычисления последовательно и в пуле: ```{code-cell} 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") ``` ```{code-cell} python !python /tmp/mp_procpool.py ``` В Лекции 5 при включённом GIL потоки не ускоряли такой цикл. Здесь вычисления могут идти параллельно, но в измерение также входят запуск воркеров, импорты, передача аргументов и результатов. На одном доступном ядре выигрыша не будет; даже на нескольких ускорение не обязано быть линейным. `map()` выдаёт результаты в порядке входных данных. Если нужна обработка по мере готовности, используем `submit()` и `as_completed()`, как с пулом потоков. Исключение из джобы возникает при получении результата через `Future.result()` или при обходе результатов `map()`. Выход из `with` ждёт завершения отправленных джоб.
Больше воркеров не всегда быстрее. Есть доступные ядра, лимиты памяти и цена обмена данными. Универсального порога вроде "джоба должна длиться не меньше 100 мс" нет: измеряем свою нагрузку, переиспользуем пул, мелкие джобы объединяем в пакеты.
### Что именно передаётся: `pickle` Пул должен перенести функцию и её аргументы в другой процесс, затем вернуть результат. Для этого он использует **сериализацию**: представляет данные последовательностью байтов. В стандартном пуле это делает `pickle`. ```{code-cell} python 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) ``` Сериализация восстанавливает данные, а не общую ссылку. Для функции `pickle` сохраняет имя модуля и имя функции, поэтому получатель должен суметь её импортировать. Покажем, как ошибка передачи появляется через `Future`: ```{code-cell} python %%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()) ``` ```{code-cell} python !python /tmp/mp_functions.py ``` У `lambda` нет подходящего имени для восстановления. Вложенные функции и локальные классы тоже обычно не подходят для стандартного `pickle`. Выносим функцию на уровень модуля и явно передаём аргументы. Аргументы и возвращаемое значение также должны поддерживать сериализацию; например, открытый файл передать как обычный аргумент нельзя. Даже при `fork` пул пересылает джобы через очередь, поэтому это ограничение сохраняется. --- ## Часть III. Откуда берётся дочерний процесс ### `spawn`: новый интерпретатор В примерах уже использовали `spawn`: запускается новый интерпретатор, импортирует главный модуль и получает функцию с аргументами. Глобальные переменные заново инициализируются при импорте; изменения, которые родитель сделал после импорта, автоматически к дочернему процессу не попадут. Новый процесс всё же получает необходимое окружение, например переменные окружения и текущий каталог. "Свежий интерпретатор" не означает "ничего не унаследовано". Цена запуска зависит в том числе от импортов программы. ### `fork`: снимок состояния родителя На POSIX можно вызвать `fork()`: ОС создаёт процесс со снимком адресного пространства родителя. В дочернем процессе продолжается только поток, который вызвал `fork`; остальные потоки туда не переходят. Память обычно копируется по принципу **копирования при записи** (copy-on-write): родитель и потомок сначала используют одни физические страницы, а при записи получают отдельные копии изменяемых страниц. Это экономит копирование при запуске, но не делает изменения Python-объектов общими. ```text Родитель: список [1, 2] Потомок: список [1, 2] | | +-------- общие страницы -------+ | запись append(3) | отдельная копия страницы ``` В снимок могут попасть и захваченные блокировки, и открытые соединения. Если блокировку держал другой поток, в дочернем процессе уже некому её отпустить. Унаследованное сетевое соединение тоже не превращается в два независимых соединения. ### `forkserver`: снимок отдельного сервера Хотим сохранить преимущества `fork`, но не снимать состояние многопоточного приложения. Для этого запускается отдельный сервер со свежим интерпретатором; по запросу он создаёт воркеров через `fork()` **из своего состояния**. ```text Родитель -- запрос --> сервер forkserver | +-- fork() --> воркер 1 +-- fork() --> воркер 2 ``` Сервер обычно однопоточный. Он может предварительно импортировать модули; если такой импорт создаёт потоки, предпосылка безопасности нарушается. Воркер получает функцию и аргументы от родителя, но не произвольные изменения его глобальных переменных. Проверка `__name__` по-прежнему нужна. ### Что выберет Python 3.14 ```{code-cell} python import multiprocessing as mp print("Способ по умолчанию:", mp.get_start_method()) print("Доступные способы:", mp.get_all_start_methods()) ``` На Windows и macOS по умолчанию используется `spawn`; на Linux и других поддерживаемых POSIX-платформах — `forkserver`. В Python 3.14 `fork` больше не является способом по умолчанию ни на одной платформе. `forkserver` доступен и на некоторых POSIX-системах помимо Linux; смотрим результат проверки. [Документация способов запуска](https://docs.python.org/3.14/library/multiprocessing.html#contexts-and-start-methods). ### Живой дедлок: `fork` после запуска потока Поток родителя захватит `threading.Lock`. Сначала дождёмся этого через `Event`, чтобы результат не зависел от случайной задержки, затем создадим процесс. В дочернем процессе попробуем захватить ту же блокировку: ```{code-cell} python %%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() ``` ```{code-cell} python !python /tmp/fork_deadlock.py fork ``` У дочернего процесса есть захваченный `lock`, но нет потока `hold_lock`, который мог бы его отпустить. Даже освобождение блокировки у родителя не исправило бы её копию у потомка. `join(timeout=2)` ограничил ожидание родителя, а не работу дочернего процесса: для остановки понадобился отдельный `terminate()` и затем `join()`. Меняем только способ запуска: ```{code-cell} python !python /tmp/fork_deadlock.py forkserver ``` Теперь воркер создаётся из состояния сервера. Захваченная блокировка родителя туда не попала; блокировка из импортированного модуля свободна. `spawn` в этом примере тоже сработает. Это не передача `threading.Lock` между процессами: для общей блокировки нужен отдельный примитив из `multiprocessing`.
Принудительное завершение может оставить очередь или общую блокировку повреждённой, а код очистки дочернего процесса не обязан выполниться. В рабочих программах сначала организуем нормальную остановку: сообщение через очередь или общий сигнал через ctx.Event().
Для переносимых примеров выбираем `spawn`, для приложения можно использовать контекст по умолчанию. В библиотеке даём вызывающему коду передать контекст; глобальный `set_start_method()` может конфликтовать с другими библиотеками. Процесс и его общие примитивы создаём из одного контекста. --- ## Часть IV. Как получить результат и обмениваться данными **IPC** (inter-process communication) — обмен между процессами. У `Process` нет аналога `Future.result()`, поэтому способ возврата результата выбираем сами. Для отдельных джоб это уже делает пул; для длительно работающих процессов пригодятся каналы и очереди. ### `Pipe`: два конца канала Одному воркеру отправим число и получим квадрат. `Pipe()` возвращает два конца двунаправленного канала: `send()` передаёт объект, `recv()` ждёт сообщение. ```{code-cell} python %%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) ``` ```{code-cell} python !python /tmp/mp_pipe.py ``` Закрываем концы, которыми больше не пользуемся. `send()` сериализует объект через `pickle`, `recv()` восстанавливает его. Разные концы можно использовать одновременно; несколько одновременных чтений или несколько записей **в один конец** без внешней синхронизации могут повредить сообщения. ### `Queue`: раздать джобы нескольким воркерам С каналом пришлось бы самому решать, какому воркеру отправить следующую джобу. Очередь позволяет нескольким процессам безопасно извлекать сообщения: каждую джобу получает один из воркеров. Интерфейс знаком по `queue.Queue`, но объекты передаются между процессами, а не по общей ссылке. ```{code-cell} python %%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() ``` ```{code-cell} python !python /tmp/mp_queue.py ``` `None` — маркер завершения (sentinel): одному воркеру он говорит выйти из цикла. Воркеров два, поэтому маркеров тоже два. Распределение джоб и порядок печати могут меняться; равную долю работы каждому воркеру очередь не обещает. `maxsize=2` ограничивает число сообщений в очереди: если воркеры не успевают, `put()` ждёт свободного места. Размер самих сообщений при этом не ограничен. Для завершения используем явный протокол, а не проверку `empty()`: состояние очереди может измениться сразу после проверки. ### `shared_memory`: большой буфер без повторной пересылки Если каждая джоба получает большой массив, сериализация и копирование через очередь могут стать дороже вычислений. Можно один раз разместить данные в **общей памяти**, а воркерам передавать только имя блока, размер и формат. Покажем механизм на одном числе: родитель создаёт блок, дочерний процесс подключается по имени и изменяет байты, родитель читает их после `join()`. ```{code-cell} python %%writefile /tmp/mp_shared_memory.py import multiprocessing as mp from multiprocessing import shared_memory import struct def increment(name: str) -> None: block = shared_memory.SharedMemory(name=name) try: value = struct.unpack_from("i", block.buf)[0] struct.pack_into("i", block.buf, 0, value + 1) finally: block.close() if __name__ == "__main__": block = shared_memory.SharedMemory(create=True, size=struct.calcsize("i")) try: struct.pack_into("i", block.buf, 0, 41) process = mp.get_context("spawn").Process(target=increment, args=(block.name,)) process.start() process.join() print("Код завершения:", process.exitcode) print("Число в общем блоке:", struct.unpack_from("i", block.buf)[0]) finally: block.close() block.unlink() ``` ```{code-cell} python !python /tmp/mp_shared_memory.py ``` `block.buf` — доступ к общим байтам. `struct` задаёт, как интерпретировать эти байты; сам блок не знает про `int`, форму массива или тип его элементов. NumPy может построить массив поверх такого буфера без копирования данных. Первоначальное размещение существующего массива в блоке всё равно требует копии. В примере запись и чтение не пересекаются благодаря `join()`. Если несколько процессов одновременно меняют общий буфер, нужны примитивы синхронизации, например `ctx.Lock()`. Раздельная память защищает только обычные локальные объекты. `close()` закрывает доступ текущего процесса, на POSIX `unlink()` удаляет блок; владелец вызывает `unlink()` один раз, когда блок больше не нужен. На Windows блок удаляется после закрытия всех его дескрипторов, а `unlink()` ничего не делает. [Подробности жизненного цикла `SharedMemory`](https://docs.python.org/3.14/library/multiprocessing.shared_memory.html#multiprocessing.shared_memory.SharedMemory). --- ## Часть V. Параллелизм без нескольких процессов ### Субинтерпретаторы: один PID, отдельные состояния Python Процессы дали независимые интерпретаторы, но пришлось платить за создание процессов и передачу данных. Можно ли получить несколько интерпретаторов внутри одного процесса? Субинтерпретатор (subinterpreter) имеет свои модули, глобальные переменные и состояние Python. [PEP 684](https://peps.python.org/pep-0684/) добавил возможность отдельного GIL для интерпретатора в CPython 3.12; в Python 3.14 [PEP 734](https://peps.python.org/pep-0734/) добавил публичный модуль `concurrent.interpreters` и `InterpreterPoolExecutor`. Сравним PID и изменяемые данные. Через `prepare_main()` передадим сериализованный список как `bytes`, а очередь для ответа создадим через `interpreters.create_queue()`: ```{code-cell} python 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() ``` PID один, но изменение списка не затронуло главный интерпретатор. Обычные изменяемые Python-объекты не становятся общими автоматически. В примере явно передали данные через `pickle`, а ответ — кортеж из поддерживаемых типов. В то же время адресное пространство процесса общее: это изоляция состояния Python, а не граница защиты, которую даёт отдельный процесс ОС. `interp.exec()` выполняет код **в текущем потоке** и ждёт завершения. Само создание интерпретаторов не создаёт параллелизм: для него нужно выполнять их в разных потоках. Готовый вариант — `InterpreterPoolExecutor`: ```{code-cell} python %%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}") ``` ```{code-cell} python !python /tmp/mp_compare.py ``` Каждый воркер этого пула — поток со своим интерпретатором и отдельным GIL, поэтому вычисления могут использовать разные ядра. Функции, аргументы и результаты `InterpreterPoolExecutor` тоже передаются через `pickle`; это не бесплатная замена пулу процессов. Не все расширения на C поддерживают несколько интерпретаторов: проверяем совместимость нужных библиотек. Авария расширения может завершить весь процесс, включая главный интерпретатор. [Документация `InterpreterPoolExecutor`](https://docs.python.org/3.14/library/concurrent.futures.html#interpreterpoolexecutor). ### Сборка без GIL: общие объекты и параллельные потоки Субинтерпретаторы разнесли состояние Python. Другой подход — разрешить потокам одного интерпретатора выполнять Python-код одновременно, сохранив общие объекты. Для этого [PEP 703](https://peps.python.org/pep-0703/) добавил сборку `free-threaded`; в Python 3.14 она получила поддерживаемый статус по [PEP 779](https://peps.python.org/pep-0779/), но остаётся отдельной сборкой. Как в Лекции 5, проверим возможности сборки и текущий режим отдельно: ```{code-cell} python import sys import sysconfig print("Сборка поддерживает отключение GIL:", bool(sysconfig.get_config_var("Py_GIL_DISABLED"))) print("GIL сейчас включён:", sys._is_gil_enabled()) ``` Для самостоятельного сравнения можно установить сборку [через `uv`](https://docs.astral.sh/uv/concepts/python-versions/#free-threaded-python) и запустить тот же файл с вычислениями: ```bash 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](https://docs.python.org/3.14/howto/free-threading-python.html). ### Что выбрать для вычислений Сравниваем CPU-bound код на Python. Если расширение на C освобождает GIL, потоки могут использовать разные ядра и в обычной сборке. | Инструмент | Python-код на разных ядрах | Изменяемые Python-объекты | Граница отдельного процесса ОС | |---|---|---|---| | `ThreadPoolExecutor`, GIL включён | нет, внутри одного интерпретатора | общие | нет | | `ProcessPoolExecutor` | да | отдельные копии, явный обмен | да | | `InterpreterPoolExecutor` | да | отдельные состояния интерпретаторов, явный обмен | нет | | `ThreadPoolExecutor`, GIL отключён | да | общие, нужна синхронизация | нет | Для обычной сборки и независимых вычислений начинаем с `ProcessPoolExecutor`. Если важны изоляция от аварий расширений и отдельные ресурсы ОС, процессы полезны даже при доступной сборке без GIL. Для субинтерпретаторов и сборки без GIL сначала проверяем совместимость зависимостей и измеряем выигрыш. Для внешней программы используем `subprocess`; её тоже можно запускать параллельно с другими. Для ожидания ввода-вывода пригодятся потоки и асинхронное выполнение — его разберём дальше в курсе. --- ## Разбираем на семинаре ### `Popen`: читать вывод до завершения программы `run()` подходит, когда нужен итоговый результат. Если программа долго работает и печатает ход обработки, хочется видеть строки сразу. `Popen` запускает процесс и возвращает управление; чтение вывода и ожидание организуем сами: ```{code-cell} python 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) ``` `-u` отключает буферизацию вывода дочернего Python: иначе строки могли бы дойти только при завершении. При выходе из `with` закрываются потоки и ожидается процесс. В отличие от `run()`, `Popen.wait(timeout=...)` при истечении таймаута сам процесс не завершает. Если отдельно читать `stdout`, а `stderr` тоже направить в `PIPE`, дочерний процесс может заблокироваться при заполнении непрочитанного канала ошибок. Когда оба вывода нужны целиком, используем `communicate()`; для единого потока можно направить `stderr=subprocess.STDOUT`. ### `multiprocessing.Pool`: другой интерфейс пула Этот интерфейс встречается в существующем коде. `map()` собирает результаты по порядку, а `imap_unordered()` выдаёт их по мере готовности: ```{code-cell} python %%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) ``` ```{code-cell} python !python /tmp/mp_pool.py ``` У `Pool` и `ProcessPoolExecutor` различается завершение: выход из `with Pool(...)` вызывает `terminate()`, поэтому незавершённая работа может быть оборвана. В этом примере результаты получены до выхода. У `ProcessPoolExecutor` выход из контекста по умолчанию ждёт отправленные джобы. ### Пакеты мелких джоб У каждого сообщения есть цена сериализации и передачи. В `ProcessPoolExecutor.map` параметр `chunksize` объединяет несколько входных элементов в одну передачу воркеру: ```python results = pool.map(cpu_task, inputs, chunksize=16) ``` Больший пакет уменьшает число сообщений, но ухудшает распределение, если джобы сильно различаются по длительности. Значение выбираем по измерениям; для `ThreadPoolExecutor` и `InterpreterPoolExecutor` этот параметр не влияет на работу. ### Что внутри `multiprocessing.Queue` `put()` сначала ждёт свободное место по счётчику, затем добавляет объект в локальный буфер. Фоновый поток отправки (feeder thread) сериализует объекты и пишет их в канал. `get()` читает из канала под блокировкой и восстанавливает объект. Фоновый поток работает на стороне **отправителя**, а не получателя. Из этого следует неочевидная ловушка: `put()` уже вернулся, но отправитель не может завершиться, пока фоновый поток не отправит буфер. Если родитель ждёт отправителя, а канал заполнен, оба останутся ждать. Покажем ошибочный порядок, затем исправим его, прочитав сообщение: ```{code-cell} python %%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() ``` ```{code-cell} python !python /tmp/mp_queue_join.py ``` Сначала освобождаем канал чтением, затем ждём отправителя. [Описание этой ловушки](https://docs.python.org/3.14/library/multiprocessing.html#programming-guidelines). `SimpleQueue` не использует фоновый поток: `put()` сам сериализует и отправляет объект, поэтому может блокироваться на записи. У него нет ограничения `maxsize` и таймаутов `put()`/`get()`, но есть `empty()`. `JoinableQueue` добавляет `task_done()` и `join()` для ожидания **обработки** сообщений, как `queue.Queue` из Лекции 5. Это другое ожидание, чем `Process.join()`: воркер может обработать все джобы и остаться работать. ### `Manager`: общий словарь через отдельный процесс Иногда нескольким процессам нужен небольшой общий словарь, а проектировать формат байтового буфера неудобно. `Manager` хранит Python-объекты в отдельном процессе и выдаёт прокси: каждый вызов обращается к тому процессу. ```{code-cell} 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)) ``` ```{code-cell} python !python /tmp/mp_manager.py ``` Это не общая ссылка на обычный `dict`: объект живёт в процессе менеджера, а аргументы и ответы передаются сообщениями. Удобно для небольших данных координации, но часто обращаться так к большим данным дорого. Последовательность операций вроде `results["count"] += 1` не становится атомарной: чтение и запись — отдельные обращения, при конкуренции нужна блокировка. `manager.Queue()` тоже возвращает прокси. Её, например, можно передавать как аргумент джобы уже работающего пула. Обычную `ctx.Queue()` передаём воркерам при создании процессов, а не произвольной джобой через `submit()`. ## Материалы - [`subprocess`: запуск внешних программ](https://docs.python.org/3.14/library/subprocess.html) - [`multiprocessing`: процессы, контексты и обмен](https://docs.python.org/3.14/library/multiprocessing.html) - [`concurrent.futures`: пулы и результаты](https://docs.python.org/3.14/library/concurrent.futures.html) - [`pickle`: сериализация объектов](https://docs.python.org/3.14/library/pickle.html) - [`shared_memory`: общие буферы](https://docs.python.org/3.14/library/multiprocessing.shared_memory.html) - [`concurrent.interpreters`: несколько интерпретаторов](https://docs.python.org/3.14/library/concurrent.interpreters.html) - [Сборка CPython без GIL](https://docs.python.org/3.14/howto/free-threading-python.html) - [PEP 684: отдельный GIL для интерпретатора](https://peps.python.org/pep-0684/) - [PEP 734: несколько интерпретаторов в стандартной библиотеке](https://peps.python.org/pep-0734/) - [PEP 703: необязательный GIL](https://peps.python.org/pep-0703/) - [PEP 779: поддерживаемый статус сборки без GIL](https://peps.python.org/pep-0779/)