ZMQ devices, ioloop and multiprocessing

Question

I am trying to apply the PUSH/PULL pattern as depicts in the following figure:

                            | PULL  ---> Send via HTTP
                            | PULL  ---> Send via HTTP
---- PUSH ----- DEVICE ---- | PULL  ---> Send via HTTP
                            | PULL  ---> Send via HTTP
                            | PULL  ---> Send via HTTP

The PUSH socket connects to the ZeroMQ device and emits the messages which are then propagated to all connected PULL sockets. What I want to achieve is a kind of parallel processing over multiple nodes in a pipeline. When processing has been done by the PULL socket, it should forward the message via HTTP to the remote endpoint.

Here is the code:

from multiprocessing import Process
import random
import time
import zmq
from zmq.devices import ProcessDevice

from zmq.eventloop import ioloop
from zmq.eventloop.zmqstream import ZMQStream

ioloop.install()


bind_in_port = 5559
bind_out_port = 5560

dev = ProcessDevice(zmq.STREAMER, zmq.PULL, zmq.PUSH)
dev.bind_in("tcp://127.0.0.1:%d" % bind_in_port)
dev.bind_out("tcp://127.0.0.1:%d" % bind_out_port)
dev.setsockopt_in(zmq.IDENTITY, b'PULL')
dev.setsockopt_out(zmq.IDENTITY, b'PUSH')
dev.start()
time.sleep(2)


def push():
    context = zmq.Context()
    socket = context.socket(zmq.PUSH)
    socket.connect("tcp://127.0.0.1:%s" % bind_in_port)
    server_id = random.randrange(1,10005)
    for i in range(5):
        print("Message %d sent" % i)
        socket.send_string("Push from %s" % server_id)


def pull():
    context = zmq.Context()
    socket = context.socket(zmq.PULL)
    socket.connect("tcp://127.0.0.1:%s" % bind_out_port)
    loop = ioloop.IOLoop.instance()

    pull_stream = ZMQStream(socket, loop)

    def on_recv(message):
        print(message)
    pull_stream.on_recv(on_recv)

    loop.start()

Process(target=push).start()

time.sleep(2)

for i in range(2):
    Process(target=pull).start()

Although the messages are correctly sent to the ZeroMQ device, I cannot see any message received - the on_recv callback is never being called.

Any help is appreciated.

Thanks

Answer 1

there's missing code above in the device init to provide a full answer. what are dev and *port ? 1 thing could be to add the sleep(1) after .connect for ports to stabilize note: no need to set identity on push/pull