Just a little Python

MongoDB Pub/Sub with Capped Collections

2013-04-05T16:58:00.000-04:00

If you've been following this blog for any length of time, you know that my NoSQL database of choice is MongoDB. One thing that MongoDB isn't known for, however, is building a publish / subscribe system. Redis, on the other hand, is known for having a high-bandwith, low-latency pub/sub protocol. One thing I've always wondered is whether I can build a similar system atop MongoDB's capped collections, and if so, what the performance would be. Read on to find out how it turned out...

Capped Collections

If you've not heard of capped collections before, they're a nice little feature of MongoDB that lets you have a high-performance circular queue. Capped collections have the following nice features:

They "remember" the insertion order of their documents
They store inserted documents in the insertion order on disk
They remove the oldest documents in the collection automatically as new documents are inserted

However, you give up some things with capped collections:

They have a fixed maximum size
You cannot shard a capped collection
Any updates to documents in a capped collection must not cause a document to grow. (i.e. not all $set operations will work, and no $push or $pushAll will)
You may not explicitly .remove() documents from a capped collection

To create a capped collection, just issue the following command (all the examples below are in Python, but you can use any driver you want including the Javascript mongo shell):

db.create_collection(
    'capped_collection',
    capped=True,
    size=size_in_bytes,     # required
    max=max_number_of_docs, # optional
    autoIndexId=False)      # optional

In the example above, I've created a collection that takes up size_in_bytes bytes on disk, will contain no more than max_number_of_docs, and which does not create an index on the _id field as would normally happen. Above, I mentioned that the capped collection remembers the insertion order of its documents. If you issue a find() with no sort specified, or with a sort of ('$natural', 1), then MongoDB will sort your result in insertion order. (($natural, -1) will likewise sort the result in reverse insertion order.) Since insertion order is the same as the on-disk ordering, these queries are extremely fast. To see this, let's create two collections, one capped and one uncapped, and fill both with small documents:

size = 100000

# Create the collection
db.create_collection(
    'capped_collection', 
    capped=True, 
    size=2**20, 
    autoIndexId=False)
db.create_collection(
    'uncapped_collection', 
    autoIndexId=False)

# Insert small documents into both
for x in range(size):
    db.capped_collection.insert({'x':x}, manipulate=False)
    db.uncapped_collection.insert({'x':x}, manipulate=False)

# Go ahead and index the 'x' field in the uncapped collection
db.uncapped_collection.ensure_index('x')

Now we can see the performance gains by executing find() on each. For this, I'll use the IPython, IPyMongo, and the magic %timeit function:

In [72] (test): %timeit x=list(db.capped_collection.find())
1000 loops, best of 3: 708 us per loop
In [73] (test): %timeit x=list(db.uncapped_collection.find().sort('x'))
1000 loops, best of 3: 912 us per loop

So we get a moderate speedup, which is nice, but not spectacular. What becomes really interesting with capped collections is that they support tailable cursors.

Tailable cursors

If you're querying a capped collection in insertion order, you can pass a special flag to find() that says that it should "follow the tail" of the collection if new documents are inserted rather than returning the result of the query on the collection at the time the query was initiated. This behavior is similar to the behavior of the Unix tail -f command, hence its name. To see this behavior, let's query our capped collection with a 'regular' cursor as well as a 'tailable' cursor. First, the 'regular' cursor:

In [76] (test): cur = db.capped_collection.find()

In [77] (test): cur.next()
       Out[77]: {u'x': 0}

In [78] (test): cur.next()
       Out[78]: {u'x': 1}

In [79] (test): db.capped_collection.insert({'y': 1})
       Out[79]: ObjectId('515f205cfb72f0385c3c2414')

In [80] (test): list(cur)
       Out[80]:
[{u'x': 2},
 ...
 {u'x': 99}]

Notice above that the document we inserted {'y': 1} is not included in the result since it was inserted after we started iterating. Now, let's try a tailable cursor:

In [81] (test): cur = db.capped_collection.find(tailable=True)

In [82] (test): cur.next()
       Out[82]: {u'x': 1}

In [83] (test): cur.next()
       Out[83]: {u'x': 2}

In [84] (test): db.capped_collection.insert({'y': 2})
       Out[84]: ObjectId('515f20ddfb72f0385c3c2415')

In [85] (test): list(cur)
       Out[85]:
[{u'x': 3},
 ...
 {u'x': 99},
 {u'_id': ObjectId('515f205cfb72f0385c3c2414'), u'y': 1},
 {u'_id': ObjectId('515f20ddfb72f0385c3c2415'), u'y': 2}]

Now we see that both the "y" document we created before as well as the one created during this cursor's iteration included in the result.

Waiting on data

While tailable cursors are nice for picking up the inserts that happened while we were iterating over the cursor, one thing that a true pub/sub system needs is low latency. Polling the collection to see if messages have been inserted is a non-starter from a latency standpoint because you have to do one of two things:

Poll continuously, using prodigious server resources
Poll intermittently, increasing latency

Tailable cursors have another option you can use to "fix" the above problems: the await_data flag. This flag tells MongoDB to actually wait a second or two on an exhausted tailable cursor to see if more data is going to be inserted. In PyMongo, the way to set this flag is quite simple:

cur = db.capped_collection.find(
    tailable=True,
    await_data=True)

Building a pub/sub system

OK, now that we have a capped collection, with tailable cursors awaiting data, how can we make this into a pub/sub system? The basic approach is:

We use a single capped collection of moderate size (let's say 32kB) for all messages
Publishing a message consists of inserting a document into this collection with the following format: { 'k': topic, 'data': data }
Subscribing to the collection is a tailable query on the collection, using a regular expression to only get the messages we're interested in.

The actual query we use is similar to the following:

def get_cursor(collection, topic_re, await_data=True):
    options = { 'tailable': True }
    if await_data:
        options['await_data'] = True
    cur = collection.find(
        { 'k': topic_re },
        **options)
    cur = cur.hint([('$natural', 1)]) # ensure we don't use any indexes
    return cur

Once we have the get_cursor function, we can do something like the following to execute the query:

import re, time
while True:
    cur = get_cursor(
        db.capped_collection, 
        re.compile('^foo'), 
        await_data=True)
    for msg in cur:
        do_something(msg)
    time.sleep(0.1)

Of course, the system above has a couple of problems:

We have to receive every message in the collection before we get to the 'end'
We have to go back to the beginning if we ever exhaust the cursor (and its await_data delay)

The way we can avoid these problems is by adding a sequence number to each message.

Sequences

"But wait," I imagine you to say, "MongoDB doesn't have an autoincrement field like MySQL! How can we generate sequences?" The answer lies in the find_and_modify() command, coupled with the $inc operator in MongoDB. To construct our sequence generator, we can use a dedicated "sequence" collection that contains nothing but counters. Each time we need a new sequence number, we perform a find_and_modify() with $inc and get the new number. The code for this turns out to be very short:

class Sequence(object):

    def __init__(self, db, name='mongotools.sequence'):
        self._db = db
        self._name = name

    def cur(self, name):
        doc = self._db[self._name].find_one({'_id': name})
        if doc is None: return 0
        return doc['value']

    def next(self, sname, inc=1):
        doc = self._db[self._name].find_and_modify(
            query={'_id': sname},
            update={'$inc': { 'value': inc } },
            upsert=True,
            new=True)
        return doc['value']

Once we have the ability to generate sequences, we can now add a sequence number to our messages on publication:

def pub(collection, sequence, key, data=None):
    doc = dict(
        ts=sequence.next(collection.name),
        k=key,
        data=data)
    collection.insert(doc, manipulate=False)

Our subscribing query, unfortunately, needs to get a bit more complicated:

def get_cursor(collection, topic_re, last_id=-1, await_data=True):
    options = { 'tailable': True }
    spec = { 
        'ts': { '$gt': last_id }, # only new messages
        'k': topic_re }
    if await_data:
        options['await_data'] = True
    cur = collection.find(spec, **options)
    cur = cur.hint([('$natural', 1)]) # ensure we don't use any indexes
    return cur

And our dispatch loop likewise must keep track of the sequence number:

import re, time
last_id = -1
while True:
    cur = get_cursor(
        db.capped_collection, 
        re.compile('^foo'), 
        await_data=True)
    for msg in cur:
        last_id = msg['ts']
        do_something(msg)
    time.sleep(0.1)

We an actually improve upon this a tiny bit by finding the ts field of the last value in the collection and using it to initialize our last_id value:

last_id = -1
cur = db.capped_collection.find().sort([('$natural', -1)])
for msg in cur:
    last_id = msg['ts']
    break
...

So we've fixed the problem of processing messages multiple times, but we still have a slow scan of the whole capped collection on startup. Can we fix this? It turns out we can, but not without questionable "magic."

Now, for some questionable magic...

You may be wondering why I would use a strange name like ts to hold a sequence number. It turns out that there is poorly documented option for cursors that we can abuse to substantially speed up the initial scan of the capped collection: the oplog_replay option. As is apparent from the name of the option, it is mainly used to replay the "oplog", that magic capped collection that makes MongoDB's replication internals work so well. The oplog uses a ts field to indicate the timestamp of a particular operation, and the oplog_replay option requires the use of a ts field in the query.

Now since oplog_replay isn't really intended to be (ab)used by us mere mortals, it's not directly exposed in the PyMongo driver. However, we can manage to get to it via some trickery:

from pymongo.cursor import _QUERY_OPTIONS

def get_cursor(collection, topic_re, last_id=-1, await_data=True):
    options = { 'tailable': True }
    spec = { 
        'ts': { '$gt': last_id }, # only new messages
        'k': topic_re }
    if await_data:
        options['await_data'] = True
    cur = collection.find(spec, **options)
    cur = cur.hint([('$natural', 1)]) # ensure we don't use any indexes
    if await:
        cur = cur.add_option(_QUERY_OPTIONS['oplog_replay'])
    return cur

(Yeah, I know it's bad to import an underscore-prefixed name from another module. But it's marginally better than simply saying oplog_replay_option=8, which is the other way to make this whole thing work....)

Performance

So now we have the skeleton of a pubsub system using capped collections. If you'd like to use it yourself, all the code is available on Github in the MongoTools project. So how does it perform? Well obviously the performance depends on the particular type of message passing you're doing. In the MongoTools project, there are a couple of Python example programs latency_test_pub.py and latency_test_sub.py in the mongotools/examples/pubsub directory that allow you to do your own benchmarking. In my personal benchmarking, running everything locally with small messages, I'm able to get about 1100 messages per second with a latency of 2.5ms (with publishing options -n 1 -c 1 -s 0), or about 33,000 messages per second with a latency of 8ms (this is with -n 100 -c 1 -s 0). For pure publishing bandwidth (the subscriber can't consume this many messages per second), I seem to max out at around 75,000 messages (inserts) per second.

So what do you think? With MongoTools pubsub module is MongoDB a viable competitor to Redis as a low-latency, high-bandwidth pub/sub channel? Let me know in the comments below!

MongoDB Schema Design at Scale

2012-09-25T12:04:00.000-04:00

I had the recent opportunity to present a talk at MongoDB Seattle on Schema Design at Scale. It's basically a short case study on what steps the MongoDB Monitoring Service (MMS) folks did to evolve their schema, along with some quantitative performance comparisons between different schemas. Given that one of my most widely read blog posts is still MongoDB's Write Lock, I decided that my blog readers would also be interested in the quantitative comparison as well.

MongoDB Monitoring Service

First off, I should mention that I am not now, nor have I ever been, an employee of 10gen, the company behind MongoDB. I am, however, a longtime MongoDB user, and have seen a lot of presentations and use cases on the database. My knowledge of MMS's internal design comes from watching publically-available talks. I don't have any inside knowledge or precise performance numbers, so I decided to do some experiments on my own to see the impact of different schema designs they might have used to build MMS.

So what is MMS, anyway? The MongoDB Monitoring Service is a free service offered by 10gen to all MongoDB users to monitor several key performance indicators on their MongoDB installations. The way it works is this:

You download a small script that you run on your own servers that will periodically upload performance statistics to MMS.
You access reports through the MMS website. You can graph per-minute performance of any of the metrics as well as see historical trends.

Eating your own dogfood

When 10gen designed MMS, they decided that it would not only be a useful service for those who have deployed MongoDB, but that it would also be a showcase of MongoDB's performance, keeping the performance graphs updated in real time across all customers and servers. To that end, they store all the performance metrics in MongoDB documents and get by on a modest (I don't know exactly how modest) cluster of MongoDB servers.

To that end, it was extremely important in the case of MMS to use the hardware they had allocated efficiently. Since this service is available for real-time reporting 24 hours per day, they had to make design the system to be responsive even under "worst-case" conditions, avoiding anything in the design that would cause an uneven performance during the day.

Building an MMS-like system

Since I don't have access to the actual MMS software, I decided to build a system that's similar to MMS. Basically, what I wanted was a MongoDB schema that would allow me to keep per-minute counters on a collection of different metrics (we could imagine something like a web page analytics system using such a schema, for example).

In order to keep everything compact, I decided to keep a day's statistics inside a single MongoDB document. The basic schema is the following:

{
    _id: "20101010/metric-1",
    metadata: {
        date: ISODate("2000-10-10T00:00:00Z"),
        metric: "metric-1" },
    daily: 5468426,
    hourly: {
        "00": 227850,
        "01": 210231,
        ...
        "23": 20457 },
    minute: {
        "0000": 3612,
        "0001": 3241,
        ...
        "1439": 2819 }
}

Here, we keep the date and metric we're storing in a "metadata" property so we can easily query it later. Note that the date and metric name are also embedded in the _id field as well (that will be important later). Actual metric data is stored in the daily, hourly, and minute properties.

Now if we want to update this document (say, to record a hit to a web page), we can use MongoDB's in-place update operators to increment the appropriate daily, hourly, and per-minute counters. To further simplify things, we'll use MongoDB's "upsert" feature to create a document if it doesn't already exist (this prevents us from having to allocate the documents ahead-of-time). The first version of our update method, then, looks like this:

def record_hit(coll, dt, measure):
    sdate = dt.strftime('%Y%m%d')
    metadata = dict(
        date=datetime.combine(
            dt.date(),
            time.min),
        measure=measure)
    id='%s/%s' % (sdate, measure)
    minute = dt.hour * 60 + dt.minute
    coll.update(
        { '_id': id, 'metadata': metadata },
        { '$inc': {
                'daily': 1,
                'hourly.%.2d' % dt.hour: 1,
                'minute.%.4d' % minute: 1 } },
        upsert=True)

To use this to record a "hit" to our website, then, we would simply call it with our collection, the current date, and the measure being updated:

>>> record_hit(db.daily_hits, datetime.utcnow(), '/path/to/my/page.html')

Measuring performance

To measure the performance of this approach, I created a 2-server cluster on Amazon EC2: one server to run MongoDB and one to run my benchmark code to do a bunch of record_hit() calls, simulating different times of day to see the performance over multiple 24-hour periods. This is what I found:

Ouch! For some reason, we see the performance of our system steadily decrease from 3000-5000 writes per second to 200-300 writes per second as the day goes on. This, it turns out, happens because our "in-place" update was not, in fact, in-place.

Growing documents

MongoDB allows you great flexibility when updating your documents, even allowing you to add new fields and cause the documents to grow in size over time. And as long as your documents don't grow too much, everything just kind of works. MongoDB will allocate some "padding" to your documents, assuming some growth, and as long as you don't outgrow your padding, there's really very little performance impact.

Once you do outgrow your padding, however, MongoDB has to move your document to another location. As your documeng gets bigger, this takes longer (more bytes to copy and all that). So documents that grow and grow and grow are a real performance-killer with MongoDB. And that's exactly what we have here. Consider the first time we call record_hit during a day. After this, the document looks like the following:

{
    _id: ...,
    metadata: {...},
    daily: 1,
    hourly: { "00": 1 }, 
    minute: { ""0000": 1 }
}

Then we record a hit during the second minute of a day and our document grows:

{
    _id: ...,
    metadata: {...},
    daily: 2,
    hourly: { "00": 2 }, 
    minute: { ""0000": 1, "0001": 1 }
}

Now, even if we're only recording a single hit per minute, our document had to grow 1439 times, and by the end of the day it takes up substantially more space than it did when we recorded our first hit just after midnight.

Fixing with preallocation

The solution to the problem of growing documents is preallocation. However, we'd prefer not to preallocate all the documents at once (this would cause a large load on the server), and we'd prefer not to manually schedule documents for preallocation throughout the day (that's just a pain). The solution that 10gen decided upon, then, was to randomly (with a small probability) preallocate tomorrow's document each time we record a hit today.

In the system I designed, this preallocation is performed at the beginning of record_hit:

def record_hit(coll, dt, measure):
    if PREALLOC and random.random() < (1.0/2000.0):
        preallocate(coll, dt + timedelta(days=1), measure)
    # ...

Our preallocate function isn't that interesting, so I'll just show the general idea here:

def preallocate(coll, dt, measure):
    metadata, id = # compute metadata and ID
    coll.update( 
       { '_id': id },
       { '$set': { 'metadata': metadata },
         '$inc': { 
             'daily': 0,
             'hourly.00': 0,
             # ...
             'hourly.23': 0,
             'minute.0000': 0,
             # ...
             'minute.1439: 0 } },
       upsert=True)

There are two important things to note here:

Our preallocate function is safe. If by some chance we call preallocate on a date/metric that already has a document, nothing changes.
Even if preallocate is never called, record_hit is still functionally correct, so we don't have to worry about the small probability that we get through a whole day without preallocating a document.

Now with these changes in place, we see much better performance:

We've actually improvide performance in two ways using this approach:

Preallocation means that our documents never grow, so they never get moved
By preallocating throughout the day, we don't have a "midnight problem" where our upserts all end up inserting a new document and increasing load on the server.

We do, however, have a curious downward trend in performance throughout the day (though much less drastic than before). Where did that come from?

MongoDB's storage format

To figure out the downward performance through the day, we need to take a brief detour into the actual format that MongoDB uses to store data on disk (and memory), BSON. Normally, we don't need to worry about it, since the pymongo driver converts everything so nicely into native Python types, but in this case BSON presents us a performance problem.

Although MongoDB documents, such as our minute embedded document, are represented in Python as a dict (which is a constant-speed lookup hash table), BSON actually stores documents as an association list. So rather than having a nice hash table for minute, we actually have something that looks more like the following:

minute = [
    [ "0000", 3612 ],
    [ "0001", 3241 ],
    # ...
    [ "1439", 2819 ] ]

Now to actually update a particular minute, the MongoDB server performs the something like the following operations (psuedocode, with lots of special cases ignored):

inc_value(minute, "1439", 1)

def inc_value(document, key, value)
    for entry in document:
        if entry[0] == key:
            entry[1] += value
            break

The performance of this algorithm, far from our nice O(1) hash table, is actually O(N) in the number of entries in the document. In the case of the minute document, MongoDB has to actually perform 1439 comparisons before it finds the appropriate slot to update.

Fixing the downward trend with hierarchy

To fix the problem, then, we need to reduce the number of comparisons MongoDB needs to do to find the right minute to increment. The way we can do this is by splitting up the minutes into hours. Our daily stats document now looks like the following:

{ _id: "20101010/metric-1",
  metadata: {
    date: ISODate("2000-10-10T00:00:00Z"),
    metric: "metric-1" },
  daily: 5468426,
  hourly: {
    "0": 227850,
    "1": 210231,
    ...
    "23": 20457 },
  minute: {
    "00": {        
        "0000": 3612,
        "0100": 3241,
        ...
    }, ...,
    "23": { ..., "1439": 2819 }
}

Our record_hit and preallocate routines have to change a bit as well:

def record_hit_hier(coll, dt, measure):
    if PREALLOC and random.random() < (1.0/1500.0):
        preallocate_hier(coll, dt + timedelta(days=1), measure)
    sdate = dt.strftime('%Y%m%d')
    metadata = dict(
        date=datetime.combine(
            dt.date(),
            time.min),
        measure=measure)
    id='%s/%s' % (sdate, measure)
    coll.update(
        { '_id': id, 'metadata': metadata },
        { '$inc': {
                'daily': 1,
                'hourly.%.2d' % dt.hour: 1,
                ('minute.%.2d.%.2d' % (dt.hour, dt.minute)): 1 } },
        upsert=True)

def preallocate(coll, dt, measure):
    '''Once again, simplified for explanatory purposes'''
    metadata, id = # compute metadata and ID
    coll.update( 
       { '_id': id },
       { '$set': { 'metadata': metadata },
         '$inc': { 
             'daily': 0,
             'hourly.00': 0,
             # ...
             'hourly.23': 0,
             'minute.00.00': 0,
             # ...
             'minute.00.59': 0,
             'minute.01.00': 0,
             # ...
             'minute.23.59': 0 } },
       upsert=True)

Once we've added the hierarchy and re-run our experiment, we get the nice, level performance we'd like to see:

Conclusion

It's always nice to see "tips and tricks" borne out through actual, quantitative results, so this was probably the most enjoyable talk I've ever put together. The things I got out of it were the following:

Growing documents is a very bad thing for performance. Avoid it if at all possible.
Awareness of the BSON specification and data representation can actually be quite useful when diagnosing performance problems.
To get the best performance out of your system, you need to actually run it (or a highly representative stand-in). Actually seeing the results of performance tweaking in graphical form is incredibly helpful in targeting your efforts.

The source code for all these updates is available in my mongodb-sdas Github repo, and I welcome any feedback either there, or here in the comments. In particular, I'd love to hear of any performance problems you've run into and how you got around them. And of course, if you've got a really perplexing problem, I'm always available for consulting by emailing me at Arborian.com.

Python and MongoDB Travel Plans

2012-09-13T16:27:00.001-04:00

It's been a little longer than normal between articles, so I wanted to let everyone know what I've been up to lately, and hope that I'll be able to see some of you at some upcoming events. So here's a summary of the things I'll be involved with over the next couple of months:

MongoDB Seattle

I have the privilege of speaking at MongoDB Seattle tomorrow, September 14, 2012. My talk is on "Schema Design at Scale", and wll have some insights based on experience 10gen gained when developing the MongoDB Monitoring Service. I'll even have some cool graphs based on experiments I did to illustrate the schema design tips I'll be giving.

MongoDB for Developers Online Training

I'm offering an 4-week online training class starting this Monday at 8pm EDT. I'll be delivering the training in 2 parts each week: a "lecture" portion on Mondays and an unstructured "office hours" on Tuesdays where I'll be available for questions. If you're interested, you can sign up by clicking the following link: Register for MongoDB for Developers. Right now the class is quite small, so I'll be able to spend a lot of time answering questions and interacting with students.

Python Training in Beijing

I'll be in Beijing, China to deliver an onsite "Fast Track to Python" class from October 5-10. The actual training is only on the 8-10, so if anyone's interested in getting together before that, I'd love to hang out in the People's Republic.

Big Dive and MongoTorino

I'm also happy to announce that I'm one of the featured instructors for the Big Dive training event in Turin, Italy this October. I'll be teaching a 2-day class on MongoDB and a 1-day class on Gevent, so if you're in northern Italy, I'd love to see you there! I'll also be at MongoTorino on October 19th, but the webpage for that doesn't seem to be updated yet. I'll be in Turin from October 13-20 if you want to come to either event, or just hang out over a bottle of wine.

PyCarolinas

To cap off my world tour I'm happy to be able to speak on Gevent and Socket.io at PyCarolinas on October 21st, so I'd love to see any of you there. Sadly I'll have to miss the 1st day of PyCarolinas on my way back from Italy.

And that's about it (and that's enough!). I'd love to see any of you who are in any of the cities mentioned above when I'm passing through, so feel free to contact me on twitter (@rick446), email (rick at arborian.com), or in the comments below!

Using ZeroMQ devices to support complex network topologies

2012-08-28T13:53:00.000-04:00

Continuing in my ZeroMQ series, today I'd like to look at ZeroMQ "devices" and how you can integrate ZeroMQ with Gevent so you can combine the easy networking topologies of ZeroMQ with the cooperative multitasking of ZeroMQ. If you're just getting started with ZeroMQ, you might want to check out the following articles:

And if you want some background on Gevent, you might want to check out that series at the following links:

Once you're caught up, let's get started...

ZeroMQ "devices"

One of the nice aspects of ZeroMQ is that it decouples the communication pattern with the connection pattern of your endpoints. Without ZeroMQ, you'd commonly have a "server" socket which binds to a port and receives requests, while "clients" connect to that socket and send requests. With ZeroMQ, it's perfectly acceptable to have the "server" connect to the "client" to receive requests (and in fact you can have multiple "servers" connected to the same "client" socket).

While this is handy, in some cases you may want to have both client and server relatively dynamic, with both connecting and neither binding to a particular port. For this use case, ZeroMQ provides so-called "devices" that bind a couple of sockets and perform forwarding operations between them to support common communication patterns. The "client" and "server" then both connect to the device. In this section, I'll cover the devices provided by the ZeroMQ library.

Queue device

The Queue device is responsible for mediating the REQ/REP communication 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))
    print 'REP is', sock.recv()

and the following response code:

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)
    print 'REP is', 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])
zmq.device(zmq.QUEUE, s1, s2)

Note that we've used a couple of new socket types above: zmq.ROUTER and zmq.DEALER. These are similar to zmq.REP and zmq.REQ, respectively, but they allow us to break the strict "request-response" lockstep of REQ/REP. The way they work is the following:

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.
The FORWARDER device sends the message received on the ROUTER to the DEALER socket.
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.
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.
The FORWARDER device sends the message received on the DEALER to the ROUTER socket.
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)

zmq_queue_device(s1, s2)

WARNING Though the code above seems to work ok, but as Dan Fairs points out, ZeroMQ sockets are not thread-safe. So what you really want to do is use nonblocking IO and zmq.Poller(), but that's a bit beyond the scope of what I wanted to cover in this article.

Forwarding device

In the same way the QUEUE device mediates the REQ/REP pattern, the FORWARDER device mediates the PUB/SUB pattern. Since PUB/SUB doesn't require the lockstep operation of REQ/REP, we can use the regular PUB/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, '')

zmq.device(zmq.FORWARDER, s1, s2)

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/SUB broker. 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, our FORWARDER device is just the following:

def zmq_forwarder_device(upstream, downstream):
    zeromq_relay(upstream, downstream)

Streaming device

Similar to the FORWARDER device, the STREAMER device just sends upstream packets downstream, but in this case in support of the PUSH/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])

zmq.device(zmq.STREAMER, s1, s2)

And once again, if we wanted to create the device manually, it's just a relay:

def zmq_streaming_device(upstream, downstream):
    zeromq_relay(upstream, downstream)

Integration with gevent

You may have noticed that the gevent device function doesn't return. If you want to create multiple devices within your Python program, then, you'll need to wrap the devices in threads.

Another approach that you might use if you, like me, prefer the lightweight threading and asynchronous I/O of gevent, is to use the gevent-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)
    sock.send(sys.argv[1] + ':' + time.ctime())

Simple, right? The reason I throw this in this seemingly-unrelated post is that if you want to use gevent, you can't use the built-in devices. This is because the built-in devices block, and they block in the ZeroMQ C library, not in Python where they could be "greened". So if you want a device with gevent, you'll have to write your own (which as, after all, pretty straightforward).

Conclusion

ZeroMQ devices provide a handy way to stitch together complex routing topologies and allow you to decouple the various components of your architecture. Though the built-in devices are quite simple, they provide insight into how you could build more complex devices yourself to fulfill the role of "broker" in a distributed architecture.

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!

ZeroMQ flow control and other options

2012-08-21T09:00:00.000-04:00

In a previous post, I provided an introduction to ZeroMQ. Continuing along with ZeroMQ, today I'd like to take a look at how you manage various "socket options" in ZeroMQ, particularly when it comes to flow control. If you've never used ZeroMQ, I recommend reading my previous post first. Once you're caught up, let's get started...

ZeroMQ "fire and forget"

One of the awesome things you may have noticed about ZeroMQ is that you can send a message without waiting for it to be received. In fact, the endpoint that's going to receive the message doesn't even need to be connected, and your application will happily act as if the message is winging its way along. While this is great for easily getting up to speed with ZeroMQ, there are some things you need to be aware of.

ZeroMQ is not magic; merely a close approximation

When you send a message in on a ZeroMQ socket, internally ZeroMQ is storing that message in an in-memory queue. So long as you don't send messages faster than whatever's downstream can read them, all is well. The problem comes when your downstream end can't process them fast enough.

Let's consider the following sender, which will send a short message as quickly as possible:

import sys
import time

import zmq

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

while True:
    sock.send(sys.argv[1] + ':' + time.ctime())

Now if we just run this without having a corresponding 'puller' socket draining the queue, we'll eat up memory as the 'pusher' just keeps enqueueing messages. On my laptop, for instance, the python process reached 3GB of virtual memory size in two minutes. Of course, in a real system you'll have a pulling socket, but in many cases your 'pulling' socket may not be able to pull has fast as your pusher can push.

High water mark to the rescue

To handle situations like this, ZeroMQ provides a socket option called a high water mark, accessed as zmq.HWM. This tells us how many messages we want ZeroMQ to buffer in RAM before blocking the 'pushing' socket. To set the high water mark, we just need to use the .setsockopt method:

...
sock = context.socket(zmq.PUSH)
sock.setsockopt(zmq.HWM, 1000)
sock.connect(sys.argv[1])

while True:
    sock.send(sys.argv[1] + ':' + time.ctime())

The modified pusher will send 1000 messages and then block, using a maximum of 2.3MB on my laptop. Note that the high water mark must be set before you connect to any clients, as ZeroMQ uses a queue-per-client, and fixes the queue size on connect.

Queueing messages on disk

There is one case in which a sending socket may exceed its high water mark. When you set the zmq.SWAP option on a socket, ZeroMQ will use a local swapfile to store messages that exceed the high water mark. In order to set up an 200KB on-disk swap file, for instance, we could use the following code:

...
sock = context.socket(zmq.PUSH)
sock.setsockopt(zmq.HWM, 1000)
sock.setsockopt(zmq.SWAP, 200*2**10)
sock.connect(sys.argv[1])

while True:
    sock.send(sys.argv[1] + ':' + time.ctime())

Lingering messages

ZeroMQ is designed to deliver messages as reliably as possible by default. One way it does this is by allowing outgoing messages to 'linger' in their queues even when the socket that sent them has been closed. For instance, suppose we have the following single-message pusher:

import sys
import time

import zmq

context = zmq.Context()
sock = context.socket(zmq.PUSH)
sock.connect(sys.argv[1])
sock.send(sys.argv[1] + ':' + time.ctime())
print 'Exiting...'

Now if we run this without a corresponding "pull" socket, our program will simply sit there saying it's exiting, but never really exiting. This is because by default, the ZeroMQ communication thread will hang around until all its outgoing messages have been sent even if the socket is closed. To modify this behavior, we can set the zmq.LINGER on the socket, setting a maximum amount of time in milliseconds that the thread will try to send messages after its socket has been closed (the default value of -1 means to linger forever):

...
sock = context.socket(zmq.PUSH)
sock.setsockopt(zmq.LINGER, 1000)
sock.connect(sys.argv[1])
sock.send(sys.argv[1] + ':' + time.ctime())
print 'Exiting...'

Other options

In the previous article, we already saw the zmq.SUBSCRIBE and zmq.UNSUBSCRIBE options. There are also a number of other socket options available for use with setsockopt. Several of these (zmq.RATE, zmq.RECOVERY_IVL,zmq.RECOVERY_IVL_MSEC, zmq.MCAST_LOOP, zmq.RECONNECT_IVL, and zmq.RECONNECT_IVL_MAX) have to do with multicast sockets (zmq.PGM and zmq.EPGM), so I won't go into them here. Others (zmq.SNDBUF, zmq.RCVBUF, and zmq.BACKLOG) have to do with the underlying OS sockets.

There is one option that's potentially a bit more interesting: zmq.IDENTITY. From the ZeroMQ docs on setsockopt:

If the socket has no identity, each run of an application is completely separate from other runs. However, with identity set the socket shall re-use any existing ØMQ infrastructure configured by the previous run(s). Thus the application may receive messages that were sent in the meantime, message queue limits shall be shared with previous run(s) and so on.

So if you create a socket and set its identity, it will pick up all the other settings you've set on it before. Beware of overusing this setting, however, since it's may be removed soon.

Conclusion

In the setsockopt method, ZeroMQ provides a way to control ZeroMQ sockets at a lower level than the "fire-and-forget" model at the higher level. Particularly if you're building a "pipeline" style application, you need to be aware of its flow control features. (I have particular experience with errors caused by not understanding the zmq.HWM option, for example.)

There is, of course, still more to cover here. In particular, I'll cover the use of devices to construct more elaborate network topologies and the use of ZeroMQ with gevent in future articles. I'd also love to hear about other topics you'd like to read about, so let me know in the comments below!