Gevent Monkeypatching ломая многопроцессорность

Question

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

Вот упрощенный фрагмент кода:

import multiprocessing

from gevent import monkey
monkey.patch_all(thread=False)

manager = multiprocessing.Manager()
q = manager.Queue()

Вот исключение, которое он создает:

Traceback (most recent call last):
  File "multimonkeytest.py", line 7, in <module>
    q = manager.Queue()
  File "/usr/local/Cellar/python/2.7.2/Frameworks/Python.framework/Versions/2.7/lib/python2.7/multiprocessing/managers.py", line 667, in temp
    token, exp = self._create(typeid, *args, **kwds)
  File "/usr/local/Cellar/python/2.7.2/Frameworks/Python.framework/Versions/2.7/lib/python2.7/multiprocessing/managers.py", line 565, in _create
    conn = self._Client(self._address, authkey=self._authkey)
  File "/usr/local/Cellar/python/2.7.2/Frameworks/Python.framework/Versions/2.7/lib/python2.7/multiprocessing/connection.py", line 175, in Client
    answer_challenge(c, authkey)
  File "/usr/local/Cellar/python/2.7.2/Frameworks/Python.framework/Versions/2.7/lib/python2.7/multiprocessing/connection.py", line 409, in answer_challenge
    message = connection.recv_bytes(256)         # reject large message
 IOError: [Errno 35] Resource temporarily unavailable

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

Если я monkeypatch внутри подпроцесса, очередь создается успешно, но когда подпроцесс пытается получить () из очереди, возникает очень похожее исключение.Сокет должен быть пропатчен из-за большого количества сетевых запросов в подпроцессах.

Моя версия gevent, которая, как мне кажется, самая последняя:

>>> gevent.version_info
(1, 0, 0, 'alpha', 3)

Есть идеи?

Answer 1

использование monkey.patch_all(thread=False, socket=False)

Я столкнулся с той же проблемой в аналогичной ситуации и отследил ее до строки 115 в gevent/monkey.py в функции patch_socket(): _socket.socket = socket.socket. Комментирование этой строки предотвращает поломку.

Здесь gevent заменяет библиотеку stdlib socket своей собственной. multiprocessing.connection довольно широко использует библиотеку socket и явно не терпит этого изменения.

В частности, вы увидите это в любом сценарии, когда импортируемый вами модуль выполняет вызов gevent.monkey.patch_all() без установки socket=False. В моем случае это был grequests, который сделал это, и мне пришлось отменить исправление модуля сокета, чтобы исправить эту ошибку.

Answer 2

К сожалению, известно, что применение многопроцессорной обработки в контексте gevent вызывает проблемы.Тем не менее, ваше обоснование разумно («много сетевой активности, но также много процессорной активности»).Если хотите, взгляните на http://gehrcke.de/gipc. Это разработано в первую очередь для вашего случая использования.С помощью gipc вы можете легко создавать несколько полностью дочерних процессов, полностью осведомленных о событиях, и позволить им совместно общаться друг с другом и / или с родителем по каналам.

Если у вас есть конкретные вопросы, вы можете их получить.назад ко мне.

Answer 3

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

import multiprocessing

from gevent import monkey
monkey.patch_all(thread=False)

q= multiprocessing.Queue()

Answer 4

Написал сменный плагин Nose Multiprocess - он должен хорошо работать со всеми сумасшедшими патчами на основе Gevent.

https://pypi.python.org/pypi/nose-gevented-multiprocess/

https://github.com/dvdotsenko/nose_gevent_multiprocess

• Переключение с multiprocess.fork на обычный subprocess.popen для рабочих процессов (для меня исправляет проблемы с ошибками общих объектов на уровне модуля)
• Переключение с multiprocess.Queue на JSON-RPC по HTTP для master-to-clientRPC
• Теперь это теоретически позволяет распределять тесты по нескольким машинам

Answer 5

Ваш предоставленный код работает для меня в Windows 7.

РЕДАКТИРОВАТЬ:

Удален предыдущий ответ, потому что я пробовал ваш код на Ubuntu 11.10 VPS, иЯ получаю ту же ошибку.

Похоже У Eventlet тоже есть эта проблема