zeromq

• Thanks for putting this up as a thread peter
• I would assume a generic persistence layer could assure reliability for one-to-one or one-to-many. In both the cases if a confirmation is not received from all receivers, it could simply resend and idempotency would be handled by the receiver as you have stated. I guess the problem then becomes how long to persist that message if acknowledgements are not received from receivers. The message could have an expiry and be deemed to be deletable if now() crosses expiry

I believe a good starting point for this discussion is the functionality offered by the JMS spec. It is well known, mostly understandable, and seems to satisfy many requirements in this space.

So, for delivery, these might be a start:

1. guaranteed once and only once
2. guaranteed at least once
3. not guaranteed at all, i.e., could be none, one, or many
4. fire and forget, i.e., no response required nor expected

The above pertain to quite possibly more than one message at a time as messages may be batched in brackets (transactions).

The client should be unaware of 'expected' problems in the infrastructure such as network down, server process overloaded, network congestion, and so forth. The client may be made aware of fatal errors, i.e., those from which no recovery is deemed possible, e.g., natural disasters. It might be feasible to include such things as full disks at this point, but that is perhaps up for debate, at least the moment when the error is returned to the client.
The same should apply to the server process when replying to messages.

I shall stop at this point as I am not sure if this is what is being asked for? Reliability is a rather loose, ill defined term, so I do not want to waste people's time with unrelated content.

As another discussion point, you could look at the QoS models for MQTT (www.mqtt.org). QoS 2 = guaranteed once and only once, QoS 1 = guaranteed at least once, QoS 0 = fire and forget.

MQTT is a lightweight protocol, and simple to encode/decode. Having said that, there are gaping holes in the semantic parts of the spec. For example:

• It defines a 16-bit message sequence number, but no 'window' or way to determine whether a particular message sequence number can be re-used or not.

• There is no flow control other than TCP blocking, and no NAK of a dropped message. Is a receiver supposed to apply TCP back pressure, which could result in deadlock, or is it supposed to drop messages it cannot handle, and rely on the sender retransmitting them periodically?

• Retrying is permitted, but the circumstances under which you may retransmit, and the algorithm for retranmission interval, is not defined. (Retransmitting if the TCP connection hasn't been dropped seems a dubious thing to do, unless the receiver is permitted to accept and discard messages when it has no space for them)

• On retransmission you are supposed to set a DUP bit - but there is no indication of how this should be processed at the receiving side

• The client can identify itself, but there is no authentication mechanism - unless you consider the client ID itself to be a plaintext password

Anyway, I'm not proposing much be taken from this, but I think it's a useful comparison point.

I'm looking for "guaranteed at least once" delivery, and some sort of lightweight persistent queue which can be embedded in the 'client' or 'source' side. Writing message(s) into a queue gets them stored to disk and Ack(s) sent when the file is flushed. When message(s) have been forwarded and the Ack(s) received, then they are freed from disk.

I think the MQTT QoS 2 model implies that "guaranteed once and only once" can be implemented as a layer on top of "guaranteed at least once".

The application I have in mind is a RADIUS server which receives RADIUS
accounting packets and forwards them to a central data warehouse for
processing.

The data warehouse and/or the intervening network connection may not be
available 100% of the time. So the RADIUS server will need to Ack the
messages back to the NAS (which can't hold them for very long), spool them
to persistent storage, and de-spool them later when available. It doesn't
need to be 100% bomb-proof but in the normal case it should survive a
process stop/start.

Another similar application might be forwarding syslogs over zeromq, where
you don't want to lose any syslog messages if the remote syslog server is
not reachable.

The following requirements come out of this.

In the API, I want to be able to send a message and then either recv() or
get a callback when the message has been successfully delivered or queued.
This is so that I can hold off sending the RADIUS Acct-Response until the
request has been safely processed or stored.

(RADIUS Accounting runs over UDP, and has a very simple assured delivery
mechanism; the NAS sends an Accounting-Request packet with an 8-bit sequence
number, and resends periodically until it gets the corresponding
Accounting-Response. See RFC 2865/2866)

(A good way to build this is OpenRADIUS, which lets you plug in an external
process over stdin/stdout to receive and response to RADIUS messages)

I think it should be possible to plug in queues where needed, without
changing the code at either end. Consider a remote SQL writer process which
returns an ACK after committing to the database. In the simplest case,
these ACKs could be used to return the RADIUS Acct-Responses to the NAS:

         +------------+           data            +----------+
NAS -----|-> RADIUS --|---------------------------|-> SQL    |--> DB
    <----|-- server <-|---------------------------|-- writer |
         +------------+           ack             +----------+
           (stateless)

Then I would like to be able to decouple them by inserting a persistent
queue process, with no code changes.

         +----------------------+                 +----------+
NAS -----|-> RADIUS --> Queue --|-----------------|-> SQL    |
    <----|-- server <-- (disk)<-|-----------------|-- writer |
         +----------------------+                 +----------+

This ensures the NAS gets its Acct-Response quickly, even if the remote
SQL writer is down or unreachable.

I might also want to insert an inbound Queue at the SQL writer side too -
for example, to be able to batch up a number of RADIUS packets into a single
SQL update.

So basically, we need the "DATA - ACK" pattern. Adding a queue means that
the protocol is unchanged but you get your ACKs back quicker.

          message              message
       --------------> QUEUE ----------->
       <--------------       <-----------
           ack(1)                ack(2)

ack(1) can be returned once the message is queued, and ack(2) will cause it
to be removed from the queue.

To get reasonable message throughput, there needs to be a number of
unacknowledged messages in transit at any one time. So the existing 0MQ
request-reply pattern may not be ideal, as it would require many concurrent
TCP connections to get that throughput.

I would like the queue process to be able to perform batching: appending
multiple messages to a file, fsyncing it after a configured time (say
100ms), and then returning acks after the fsync. This gives a potential
throughput of tens of thousands of messages per second even on a single
commodity hard drive (M10K), depending on what latency tradeoff you're
prepared to accept.

There is an even bigger potential benefit from this batching. If the
message is forwarded before it is written to disk, and the ack comes back
before the timeout expires, then there is no need to write it to disk at
all! You thus get the benefit of persistent storage without any performance
penalty in the usual case.

Ideally I'd like to get this benefit if the queues are chained too. The
message should propagate as far as it can and only get written to disk at
the last point. This may be awkward to arrange if each queue has its own
independent timeout.

Sender ----------> QUEUE ----------> QUEUE ---------> receiver ---> SQL
       <----------       <----------       <--------- process
           ack               ack               ack

If you have a queue feeding multiple recipients in a round-robin way, you
want to have some decent flow control - or be able to configure a window
size of 1 message.

                   ,------> rx1
    -------> QUEUE -------> rx2
                   `------> rx3

Consider that if you commit a large burst of messages to rx1, rx2 and rx3,
it might be that rx2 and rx3 become idle while rx1 is still working through
its backlog.

In this application I don't actually care about in-order delivery, but
it may be simplest to implement it as such. Then all you need is a TCP-like
message sequence number and ack number, and you get duplicate-prevention
too.

However, this becomes complicated if there are multiple concurrent
connections between the same client-server pair, or if routing different
messages to different destinations (see rx1,rx2,rx3 above) which may take
varying times to process each message.

So perhaps it is better to acknowledge each message individually, like
RADIUS and MQTT. What is the scope of the message ID? Should it be globally
unique, or unique for the particular (client IP,server IP,server port)?

Other thoughts.

• Do we want an API to poll for messages in a queue, or should a queue always "push" messages to the recipient(s)?

• We may want to attach a number of queues to a single pub/sub stream, each with its own filter.

                    +-- Q1 -->
   publish -------->+-- Q2 -->
                    +-- Q3 -->

Similarly, the output of a queue may be a publish stream.

                    +-- sub1
       Q1 --------->+-- sub2
                    +-- sub3

Not done it for zeromq, but for quite a few others to make them work at some reasonable speed without disk e.g. for large grid calc engines which tend to be unrelaible.

At the application node level, the queue isn't be a queue in the normal sense, it is a historical state machine modelling a distributed journal of event notifications. Nothing gets popped, just its status is marked until some archival process comes along and sweeps away old stuff.

Reliability is separated into confirmation (normal reliability) and recovery (node failure), much like clustered session state.

Both can be seen as the detection of, and querying from upstream for missed messages.

An important concept that is needed is the ability to send null messages for confirmation of reachability both upstream and downstream (heartbeats).

This is all a lot of work for the messaging layer, and starts to turn it into something else : a fully distributed transaction manager.

May I assume that this topic is not about 0MQ functionality per se?

0MQ remains 'sockets on steroids' and provides a highly efficient, easy to program (sockets) API and is as 'reliable' as the transport method it uses, e.g., TCP/IP?

Anyone wishing additional reliability, be it durable messages or whatever, implements that on top of the 0MQ layer (SPB framing)?

Apologies for perhaps stating the obvious and partly repeating the introduction to the topic!

Is it possible to setup 0MQ such that the sender/publisher publishes to "multiple" 0MQ nodes which are fanned in for the receiver/subscriber? In other words, sender -> n x (0MQ nodes) -> fan in -> receiver. This way, if one of the nodes (actually n-1 nodes) dies, the message can still have guaranteed delivery. A simple replication-based scheme maybe better in terms of performance for message durability than true persistence, right?

Has there been any movement over the last year regarding reliability under 0MQ?