Using ZeroMQ Devices to Support Complex Network Topologies

• Introduction to ZeroMQ
• ZeroMQ flow control and other options

• Introduction to Gevent
• Gevent, Threads, and Benchmarks
• Gevent and Greenlets
• Greening the Python Standard Library with Gevent
• Building TCP Servers with Gevent
• Building Web Applications with Gevent's WSGI Server

ZeroMQ "devices"

Queue device

The Queue device is responsible for mediating the REQ/REPcommunication pattern. Suppose we have the following request code:

import sys
import zmq

context = zmq.Context()

for x in xrange(10):
    sock = context.socket(zmq.REQ)
    sock.connect(sys.argv[1])
    print 'REQ is', x,
    sock.send(str(x)) 

import sys
import zmq

context = zmq.Context()

while True:
    sock = context.socket(zmq.REP)
    sock.connect(sys.argv[1])
    x = sock.recv()
    print 'REQ is', x,
    reply = 'x-%s' % x
    sock.send(reply) 

To set up a broker that forwards between the two, you need a tiny script: 

import sys
import zmq

context = zmq.Context()

s1 = context.socket(zmq.ROUTER)
s2 = context.socket(zmq.DEALER)
s1.bind(sys.argv[1])
s2.bind(sys.argv[2]) 

1. The ROUTER socket receives a message on a particular connection. It adds a message ID to the beginning of the message that identifies the sending REQ socket.
2. The FORWARDER device sends the message received on theROUTER to the DEALER socket.
3. The DEALER picks a connection to send the message to, stripping the prefix but noting the message ID and linking it to the REP socket handling the message.
4. The DEALER receives the response message, pairs it up with the message ID it's responding to, and adds the message ID to the beginning of the mssage.
5. The FORWARDER device sends the message received on theDEALER to the ROUTER socket.
6. The ROUTER socket strips the message ID and sends the message to the REQ socket that sent the initial request.

By using message IDs as above, we can have multiple messages 'in flight', being handled by various servers. If you'd rather not use the built-in ZeroMQ device, you can build your own fairly simply. The code below shows a device that that uses a pair of threads to relay messages from one socket to another:

import sys
import zmq
import time

context = zmq.Context()

s1 = context.socket(zmq.ROUTER)
s2 = context.socket(zmq.DEALER)
s1.bind(sys.argv[1])
s2.bind(sys.argv[2])

def zeromq_relay(a, b):
    '''Copy data from zeromq socket a to zeromq socket b'''
    while True:
        msg = a.recv()
        more = a.getsockopt(zmq.RCVMORE)
        if more:
            b.send(msg, zmq.SNDMORE)
        else:
            b.send(msg)

def zmq_queue_device(s1, s2):
    import threading
    t1 = threading.Thread(target=zeromq_relay, args=(s1,s2))
    t2 = threading.Thread(target=zeromq_relay, args=(s2,s1))
    t1.daemon = t2.daemon = True
    t1.start()
    t2.start()
    while True:
        time.sleep(10)
 

Forwarding device

In the same way the QUEUE device mediates the REQ/REP pattern, theFORWARDER device mediates the PUB/SUB pattern. Since PUB/SUB doesn't require the lockstep operation of REQ/REP, we can use the regularPUB/SUB sockets in our device: 

import sys
import zmq
context = zmq.Context()

s1 = context.socket(zmq.SUB)
s2 = context.socket(zmq.PUB)

s1.bind(sys.argv[1])
s2.bind(sys.argv[2])

s1.setsockopt(zmq.SUBSCRIBE, '')
 

Now we can connect one or more publishers to our device's "upstream" port (sys.argv[1]), and one or more subscribers to the device's "downstream" port (sys.argv[2]) to provide a PUB/SUBbroker. Note in particular that we had to subscribe to all mesages in the device code since we don't know which messages our downstream sockets are interested in. 
If you'd rather filter messages that get fowarded, you can subscribe to some subset of messages. Unfortunately, I'm not aware of any way to forward only messages that the downstream clients have subscribed to using built-in ZeroMQ functionality.
If we want to write our own FORWARDER device, it's even simpler than the QUEUE device since it only handles unidirectional communication. Assuming we have the zeromq_relay function as defined above, ourFORWARDER device is just the following:

def zmq_forwarder_device(upstream, downstream): 

Streaming device

Similar to the FORWARDER device, the STREAMER device just sends upstream packets downstream, but in this case in support of thePUSH/PULL pattern rather than PUB/SUB. To make a STREAMER broker, then, we just need the following code:  

import sys
import zmq
context = zmq.Context()

s1 = context.socket(zmq.PULL)
s2 = context.socket(zmq.PUSH)

s1.bind(sys.argv[1])
s2.bind(sys.argv[2])
 

def zmq_streaming_device(upstream, downstream): 

Integration with gevent

Another approach that you might use if you, like me, prefer the lightweight threading and asynchronous I/O of gevent, is to use thegevent-zeromq package available from PyPI: 

$ pip install gevent-zeromq  

 Now we can use a 'green' version of ZeroMQ by just importing from the gevent-zeromq wrapper in our scripts. A "pusher" script would then look like this:

import sys
import time

from gevent_zeromq import zmq

context = zmq.Context.instance()
sock = context.socket(zmq.PUSH)
sock.connect(sys.argv[1])

while True:
    time.sleep(1) 

Conclusion

I'd be interested in hearing from you how you are using devices (or whether you find them useful at all) in ZeroMQ programming. Do you use the built-in devices? Any devices you write yourself? Do you prefer the multithreaded device approach I've used here, or using the zmq.Poller object? Let me know in the comments below!