Monthly Archives: June 2011

ZeroMQ is a Message Queue (MQ) framework that plainly works. Two of the most interesting elements are 1) ZMQ supports a number of client languages; 2) the broker (generally an application that exists to route traffic from a producer to a consumer) is left to the programmer to implement.

If you’ve read or used any of the other MQs or if you’ve done some interprocess communication (IPC) before you probably have a good or general idea how this is supposed to work. RabbitMQ does a really good job naming the different patterns and keeping the list to something manageable. While the ZMQ doc is long, detailed, absent of examples for each of the client languages, examples are buggy or old, the examples are simple; but they have many more patterns than RMQ.

One idea that keeps getting trapped in my head is; How do I send a request to the broker and wait for a response. And if the response does not arrive, then what? Basically I’m looking for a best practice here.

In my application design I have the following stack:

json_rpc -> broker -> worker -> remote_http

1. a user application sends my stack a json-rpc call
2. the call is forwarded to the broker
3. the broker routes the request to a worker
4. the worker forwards the request to a remote service provider
5. and whatever response is returned… bubbles up through the call stack

So in my json-rpc application I have a module that waits for requests, the message is authenticated, and the input data is validated. Since the response time can be between 250ms and 90secs we’ll keep the socket open and wait for the reply. The challenge here is getting the json-rps app to make the request, detect errors, handle certain errors, and forward the request to the broker, wait for a reply, parse the response message, and return to the caller.

retry = 3
zmq.send(socket, request, zmq.NOBLOCK)
while True:
    # wait 7 seconds between boll timeouts
    socks = dict(poller.poll(7000))
    try:
        if socks.get(service) == zmq.POLLIN:
            # we got a message
            # NOBLOCK here almost makes no sense other than we want to get
            # all errors and we do not want to block at all
            # any errors will generate an exception
            # we could assert against an empty response but why?
            rsp = zmq.recv(socket, zmq.NOBLOCK) 
        else:
            # timeout
            retry = retry - 1
    except zmq.core.error.ZMQError, e:
        retry = retry - 1
        if zmq.EFSM == e.errno:
            # wrong state
            pass
        elif zmq.EAGAIN == e.errno
            # no data
            pass
        else:
            retry = 0
        if retry <= 0:
            # reconnect to the broker
            service.close()
            . . .
            retry = 3
            continue

The thing to keep in mind is that ZMQ will not return from a send or receive unless something good happened. That means that you can expect an exception to be fired if your message did not get out, the response was not received, some other app in the stack restarted and thus changed the state of the socket. This makes sending messages reliably very simple… even though my post is even simpler than than.

I keep keep going back and forth on the whole idea of NoSQL and it bothers me to no end. On the one side the idea of sharding the data at the server level is appealing. Then there are the Key/Value databases and then the document versions. They are graph databases, object databases and a few in between. But then, as always, there is this reality check.

As wonderful as NoSQL would appear to be there is no single use-case that would seem to make it/them the obvious choice. And this is no more obvious as I stumble around a new project I’m considering… an open source merchant gateway and an open source issuing system. I would really like to have one and only one system that I could use for everything but that does not seem possible.

For example; in order to deal with the protocol impedance I need a fifo queue of some kind. I like redis for this as it also has a TTL so that old data is simply removed from the queue. It also has the notion of fields in the value so that a single record can actually represent a dictionary or an array; making a useful container for the message components. (converting an ISO-8583 or variables from a POST into a dictionary is fun and useful). Also, since there is a fifo queue the transaction results and logging can be pushed into a different channel for aggregation and processing later.

In both the issuing and gateway systems the transactional part is considered OLTP system and OLTP systems benefit greatly from hash tables like those that redis provides. However, redis is useless when it comes to reporting, mapreduce, partitioning, and performance when the dataset approaches available memory limits. So it would seem that it would be useful to have a SQL or other storage mechanism to store all the persistant data, a queue for all of the logs and transactions, and a hash to act as a cache for the data from the primary store. The challenge here is that the first time there is a miss on the account data on the cache the compute node will have to go to storage to get the account data. This can be costly and 2x machines will negatively effect the sigma score. And then system recovery is going to be harder even if the data is replicated around the systems. (expire/TTL does not work as expected on replicated systems.)

The NoSQL Databases website claims that there are 122+ NoSQL databases. I just ran over a bunch of them and they all left me wanting something better… or at least feeling that I needed to go back to an RDBMS. (read PostgreSQL)

Some requirements:

• a queue for message logging that is persistant and replicated
• a queue for transaction logging account info back to the primary storage
• a queue for transaction requests that can and will expire, however, the expiration will trigger a log entry
• a cache that represents the account data from the primary storage
• a cache that represents the config data from the primary storage that is reloaded on-demand
• flush from cache when the transaction entry is recovered from the transaction log
• evaluate the mapreduce hit/miss when the transaction log is processed or mapreduce once a day or hour
• import the data from the cache into a detached slice until the data is ready for consumption. Then attach it. (Postgres)

I’m going to build a test harness made from redis+mongodb and redis+postgres… just for fun and testing.