Gevent monopatching rompiendo multiprocesamiento

Question

Estoy intentando utilizar el grupo de multiprocesamiento para ejecutar un grupo de procesos, cada uno de los cuales ejecutará un conjunto gevent de áreas verdes. La razón de esto es que hay mucha actividad de red, pero también mucha actividad de la CPU, así que para maximizar mi ancho de banda y todos los núcleos de mi CPU, necesito múltiples procesos y el parche mono asíncrono de gevent. Estoy usando el administrador de multiprocesamiento para crear una cola a la que accederán los procesos para que los datos se procesen.

Aquí es un fragmento simplificada del código:

import multiprocessing 

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

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

Aquí es la excepción que produce:

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 

Creo que esto debe ser debido a alguna diferencia entre el comportamiento de la toma de corriente normal, Módulo y módulo de socket gevent.

Si monopatch dentro del subproceso, la cola se crea con éxito, pero cuando el subproceso intenta obtener() de la cola, se produce una excepción muy similar. El socket necesita ser parcheado debido a la gran cantidad de solicitudes de red en los subprocesos.

Mi versión de GEvent, que creo que es la última:

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

¿Alguna idea?

Answer 1

Su código proporcionado a mí me funciona en Windows 7.

EDIT:

Eliminado respuesta anterior, porque he intentado su código en Ubuntu 11.10 VPS, y yo estoy recibiendo el mismo error .

es como Eventlet have this issue too

Answer 2

uso monkey.patch_all(thread=False, socket=False)

se han topado con el mismo problema en una situación similar y rastreado este hasta la línea 115 en gevent/monkey.py bajo la función patch_socket() Look: _socket.socket = socket.socket. Comentando esta línea evita la rotura.

Aquí es donde gevent reemplaza la biblioteca stdlib socket con la suya propia. multiprocessing.connection utiliza la biblioteca socket bastante extensamente, y aparentemente no tolera este cambio.

Específicamente, verá esto en cualquier escenario en el que un módulo que importe realice una llamada gevent.monkey.patch_all() sin establecer socket=False. En mi caso fue grequests que hizo esto, y tuve que anular el parcheo del módulo de socket para corregir este error.

Answer 3

La aplicación del multiprocesamiento en el contexto de gevent desafortunadamente se sabe que plantea problemas. Su fundamento, sin embargo, es razonable ("mucha actividad de red, pero también mucha actividad de CPU"). Si lo desea, eche un vistazo a http://gehrcke.de/gipc. Esto está diseñado principalmente para su caso de uso. Con gipc, puede generar fácilmente algunos procesos hijos totalmente conscientes del factor gevent y permitirles comunicarse entre sí y/o con el padre a través de tuberías.

Si tiene alguna pregunta específica, puede contactarme.

Answer 4

Si va a utilizar Cola original, entonces su código funcionará normalmente incluso con el zócalo mono parcheado.

import multiprocessing 

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

q= multiprocessing.Queue() 

Answer 5

Has escrito un reemplazo de la nariz multiproceso plug-in - éste debe jugar bien con todo tipo de reparación a base de GEvent loco.

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

https://github.com/dvdotsenko/nose_gevent_multiprocess

• Cambia de multiprocess.fork a llanura subprocess.popen para procesos de trabajo (arreglos módulo de nivel compartido erróneamente cuestiones objetos para mi)
• cambiaron de multiprocess.Queue a JSON-RPC a través de HTTP para maestro a clientes RPC
• Esto ahora puede permitir teóricamente que las pruebas se distribuyan a varias máquinas