Python ZeroMQ PUSH/PULL - ¿Perdió mensajes?

Question

Estoy tratando de usar python con zeroMQ en el modo PUSH/PULL, enviando mensajes de tamaño 4 [MB] cada pocos segundos.

Por alguna razón, aunque parece que se envían todos los mensajes, SÓLO ALGUNOS parecen haber sido recibidos por el servidor. ¿Que me estoy perdiendo aqui?

Aquí está el código para el cliente - client.py

import zmq 
import struct 

# define a string of size 4[MB] 
msgToSend = struct.pack('i', 45) * 1000 * 1000 

context = zmq.Context() 
socket = context.socket(zmq.PUSH) 
socket.connect("tcp://127.0.0.1:5000") 

# print the message size in bytes 
print len(msgToSend) 

socket.send(msgToSend) 

print "Sent message" 

Y aquí está el código para el servidor - server.py

import zmq 
import struct 

context = zmq.Context() 
socket = context.socket(zmq.PULL) 
socket.bind("tcp://127.0.0.1:5000") 

while True: 
    # receive the message 
    msg = socket.recv() 

    print "Message Size is: {0} [MB]".format(len(msg)/(1000 * 1000)) 

¿Qué me falta? ¿Cómo puedo garantizar que los mensajes siempre se envíen y no se pierdan?

En caso de que importe, estoy usando Ubuntu 10.04 32bit, máquina Core Duo con 2 [GB] RAM.

NOTA: me trataron el mismo ejemplo usando RabbitMQ y todo funciona bien - no hay mensaje se pierde. Estoy perplejo ya que a menudo escucho las alabanzas de zeroMQ. ¿Por qué falló cuando RabbitMQ tuvo éxito?

Answer 1

El problema es que cuando el programa sale, el socket se cierra inmediatamente y la basura se recoge con un LINGER efectivo de 0 (es decir, arroja los mensajes no enviados). Este es un problema para los mensajes más grandes porque tardan más en enviarse de lo que se necesita para que el socket sea basura.

Puede evitar esto poniendo sleep(0.1) justo antes de que el programa salga (para retrasar el socket y el contexto que es basura recolectada).

socket.setsockopt(zmq.LINGER, -1) (que es el valor predeterminado) debería evitar este problema, pero no por alguna razón no he tenido tiempo de investigar.

Answer 2

Es posible que se esté quedando sin memoria (dependiendo de cómo esté enviando mensajes, si se están consumiendo lo suficientemente rápido, etc.). Puede usar socket.setsockopt(zmq.HWM) para establecer HWM en un valor razonable y evitar que zeromq almacene demasiados mensajes en el búfer de salida. Con esto en mente, considere ejemplos ligeramente modificados:

# server 
... 
counter = 0 
while True: 
    ...receive the message 
    counter += 1 
    print "Total messages recieved: {0}".format(counter) 

# client 
socket.setsockopt(zmq.HWM, 8) 
for i in range(1000): 
    socket.send(msgToSend) 

Y luego correr 10 clientes de prueba:

for i in {1..10}; do 
    python client.py & 
done 

Desde el servidor se puede ver todos los mensajes son recibimos:

Total messages recieved: 9998 
Total messages recieved: 9999 
Total messages recieved: 10000