Многопроцессорность Python. Очередь помещает процесс уничтожения в контейнер Docker

Я провожу небольшой multiprocessing тест в контейнере Docker. В тесте используется multiprocessing.Queue с потребителем и производителем. Минимальный код, который приводит к сбою, можно найти ниже вместе с файлом Dockerfile. В коде основной процесс запускает новый процесс, который получает ссылку на Queue и использует Queue.put() для помещения целого числа в очередь к основному процессу. Я вижу, что процесс-производитель завершается во время вызова .put(), Exception вообще не вызывается.

Есть идеи, почему .put() убивает процесс? Это работает локально (macOS), но не работает с базовым образом Python для контейнера. Версия Python — 3.9.16. Обновление: это также не работает на обычной виртуальной машине Debian.

import multiprocessing as mp
import time
import traceback
from typing import Any, Optional

import psutil

base_file = "logs.txt"


def main() -> None:
    queue: Any = mp.Queue()
    print("Queue created")
    print("Starting producer process")

    p = mp.get_context("spawn").Process(target=producer, args=(queue,), daemon=True)
    p.start()
    print(f"Main: producer started: {p.pid}")

    alive = True
    while alive:
        alive = p.is_alive()
        print(f"Ha ha ha staying alive, producer: {p.is_alive()}")
        time.sleep(1)

    print("Every process is dead :( ")

def producer(q: mp.Queue) -> None:
    with open(f"producer.{base_file}", "w") as f:
        print("Producer: started", file=f, flush=True)
        current_value: int = 0
        while True:
            print(f"Producer: Adding value {current_value} to queue", file=f, flush=True)
            try:
                q.put(current_value, block=False)
            except BaseException as e:
                print(f"Producer: exception: {e}", file=f, flush=True)
                print(f"{traceback.format_exc()}", file=f, flush=True)
                raise e
            print(f"Producer: Value {current_value} added to queue", file=f, flush=True)
            print("Producer: Sleeping for 1 second", file=f, flush=True)
            time.sleep(1)
            current_value += 1


if __name__ == "__main__":
    main()
FROM python:3.9.16

RUN apt-get update && apt-get install -y gettext git mime-support && apt-get clean

RUN python3 -m pip install psutil

COPY ./multiprocessing_e2e.py /src/multiprocessing_e2e.py

WORKDIR /src

CMD ["python", "-u", "multiprocessing_e2e.py"]

Можете ли вы добавить команды сборки и запуска, включая, в частности, вывод команды запуска?

Ulrich Eckhardt 25.06.2024 16:47

Команда сборки — docker build -t mp_test ., а команда запуска — docker run -v $(pwd):/src mp_test Полученный результат в файле журнала следующий: ``` Производитель: запущен Производитель: добавление значения 0 в очередь ``` После чего процесс-производитель завершает работу

brvh 25.06.2024 16:54

Обновление: я воспроизвел поведение при запуске кода Python на простой виртуальной машине Debian (без контейнера), используя debian-12. Я нахожу это довольно странным, поскольку это самый простой пример в документации Python.

brvh 25.06.2024 16:56

Еще одно обновление - вызов queue.get_nowait() тоже вылетает

brvh 25.06.2024 16:58
отредактируйте вопрос, включив в него эту информацию, пожалуйста.
Ulrich Eckhardt 25.06.2024 17:23

ставьте все обновления под вопрос, а не в комментарии. Они будут более читабельными, и их смогут увидеть больше людей (люди могут не читать комментарии).

furas 25.06.2024 17:46
Почему в Python есть оператор "pass"?
Почему в Python есть оператор "pass"?
Оператор pass в Python - это простая концепция, которую могут быстро освоить даже новички без опыта программирования.
Некоторые методы, о которых вы не знали, что они существуют в Python
Некоторые методы, о которых вы не знали, что они существуют в Python
Python - самый известный и самый простой в изучении язык в наши дни. Имея широкий спектр применения в области машинного обучения, Data Science,...
Основы Python Часть I
Основы Python Часть I
Вы когда-нибудь задумывались, почему в программах на Python вы видите приведенный ниже код?
LeetCode - 1579. Удаление максимального числа ребер для сохранения полной проходимости графа
LeetCode - 1579. Удаление максимального числа ребер для сохранения полной проходимости графа
Алиса и Боб имеют неориентированный граф из n узлов и трех типов ребер:
Оптимизация кода с помощью тернарного оператора Python
Оптимизация кода с помощью тернарного оператора Python
И последнее, что мы хотели бы показать вам, прежде чем двигаться дальше, это
Советы по эффективной веб-разработке с помощью Python
Советы по эффективной веб-разработке с помощью Python
Как веб-разработчик, Python может стать мощным инструментом для создания эффективных и масштабируемых веб-приложений.
1
6
51
1
Перейти к ответу Данный вопрос помечен как решенный

Ответы 1

Ответ принят как подходящий

При попытке запустить ваш пример в Fedora39, Python 3.12.4, WSL я получил сообщение об ошибке, которое, вероятно, объясняет проблему:

    raise RuntimeError('A SemLock created in a fork context is being '
RuntimeError: A SemLock created in a fork context is being shared with a process in a spawn context. This is not supported. Please use the same context to create multiprocessing objects and Process.

И почему не вылетает на MacOS, так как по умолчанию используется Spawn. Вероятно, ошибочное поведение было обнаружено между Python3.9 и 3.12 и исправлено, чтобы вызвать правильную раннюю ошибку, прежде чем кто-либо попытается фактически использовать очередь, а не перенесено обратно в 3.9.

Действительно, изменение очереди и процесса для запуска из контекста «порождения» работает:

import multiprocessing as mp
import time
import traceback
from typing import Any, Optional

import psutil

base_file = "logs.txt"


def main() -> None:
    mp_ctx = mp.get_context("spawn")
    queue: Any = mp_ctx.Queue()
    print("Queue created")
    print("Starting producer process")

    p = mp_ctx.Process(target=producer, args=(queue,), daemon=True)
    p.start()
    print(f"Main: producer started: {p.pid}")

    alive = True
    while alive:
        alive = p.is_alive()
        print(f"Ha ha ha staying alive, producer: {p.is_alive()}")
        time.sleep(1)

    print("Every process is dead :( ")

...

(Кстати, spawn, скорее всего, станет поведением многопоточной обработки по умолчанию даже в Linux, поскольку недавно люди обнаружили некоторые трудные для отладки условия гонки, которые возникают в многопоточных программах, которые разветвляются)

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

brvh 02.07.2024 15:24

Другие вопросы по теме