A tour of CPPZMQ, the C++ bindings to libzmq

In this document, I tour the main parts of the cppzmq package so I can learn them better. I write down what I learn along the way so maybe others can learn something too.

Brevity is attempted and so the general concepts of ZeroMQ are assumed to already be understood. If you, the curious reader, have not yet done so please do yourself a big favor and read the entire ZeroMQ Guide. Read it at least twice. I think gets even better each time I read it as I find something new and useful that I had previously missed.

This tour may not be needed by everyone. We may make effective use of cppzmq simply by reading its header files and maybe checking the unit tests. In some cases we may also need to refer to the libzmq API documentation which should be installed on your system as Unix man pages (man zmq to start) and which are available online. This document attempts to distill what was learned in the process of just such a reading. I hope it may serve as a gentle narrative or tour that will help others quickly feel comfortable developing with cppzmq.

The cppzmq package core is a header-only library providing the C++ namespace zmq::. Currently the library consists of two headers and here is a summary of what they provide:

zmq.hpp

(single part) message, context, buffer, socket, monitor, poller

zmq_addon.hpp

multipart and related functions and another form of poller

Most of the goodies are in zmq.hpp and we go through them in order they appear in that file. The goodies in zmq_addon.hpp are just as good and I feel there's no need for a separation. All my cppzmq projects benefit from both.

The package also provides substantial unit tests which can be mined for programming examples. The package README file gives instructions for formal installation but one very simple option is copying the zmq.hpp and zmq_addon.hpp files right into our own package.

Here are some examples of constructing a message_t:

// Default constructs an empty message of size zero.
zmq::message_t msg;

// Have message allocate some memory internally.
const size_t size = 1024;
zmq::message_t msg(size);

// Initialize internal memory with a copy of external data.
const char bytes[s] = {0};
zmq::message_t msg(bytes, size);

// Or, from string literal (see also str_buffer).
zmq::message_t msg("hello world!", 12);

// Initialize through iteration on a container.
std::vector<int> ivec = {1,2,3};
zmq::message_t msg(ivec.begin(), ivec.end());
// or more simply
zmq::message_t msg(ivec);

// Zero-copy provision with custom deallocation callback.
// Wraps zmq_msg_init_data(3), see man page for details.
zmq::message_t msg(byte, size, myfree, nullptr);

// Move/swap the data of one message into a new one
zmq::message_t msg2(msg);
// or
zmq::message_t msg2 = msg;

We can also "rebuild" an existing message. The methods to do this mirror the constructor prototypes:

// Empty the message
msg.rebuild();

// Allocate a new size
msg.rebuild(size);

// Allocate and set with a copy
msg.rebuild(bytes, size);

Not included here is a zero-copy version similar to the zero-copy constructor shown above.

The message is a block of bytes and we can get at those bytes in a few ways:

// Size of the data in bytes 
size_t msg.size();

// Low level, type free access
const void* vptr = msg.data();

// As above but type cast
const int* iptr = msg.data<int>();

// With a copy into a non-const collection
std::vector ints(msg.data<int>(), msg.size()/sizeof(int));

// If the data makes sense as a string.
std::string str = msg.to_string();

// And we can get a zero copy view.
std::string_view strv = msg.to_string_view();

// An artistic string representation of the contents.
std::cout << msg.str() << std::endl;

Here we have a few details that are specific to certain socket types. We still haven't gotten to sockets, but they are coming. You may feel comfortable to skip this section for now.

There are two variants of buffers, mutable_buffer and const_buffer. As the name suggests, the data given to the first may be modified while the data given to the second may not be.

Either type of buffer may be constructed directly as in this example:

// Empty 
mutable_buffer mbuf();
const_buffer cbuf();

// Fodder data
const size_t size = 1024;
void* ptr = malloc(size);
const void* cptr = ptr;

// With data
mutable_buffer mbuf(ptr, size);
const_buffer cbuf(cptr, size);

We are also given a variety of functions named buffer() to construct buffers in useful ways. We give some examples next and will see later some examples of how to use buffers in the sections Socket and Multipart.

// Fodder data from previous example

// Basic construction
zmq::mutable_buffer mbuf = zmq::buffer(ptr, size);
zmq::const_buffer cbuf = zmq::buffer(cptr, size);

// C array. 
int data[1024];
mbuf = zmq::buffer(data);

// C++ vector, std::array is similar
std::vector<int> data(size);
mbuf = zmq::buffer(data);

// C++ string and string literal
std::string str = "hello world";
mbuf = zmq::buffer(str);
cbuf = zmq::str_buffer("hello world");

Once constructed, buffers are rather simple but we can operate on them to narrow their view of the underlying data.

// Truncate tail half, same works with const_buffer
mbuf = zmq::buffer(mbuf, mbuf.size()/2);
// Truncate front half, etc cbuf.
mbuf += mbuf.size()/2;

Why do this narrowing? One very useful pattern is for an application to take some action based on the prefix or postfix of a message. This narrowing can be performed after this information is used and the remainder can be forwarded to an output socket. No copying needed.

Unlike a message, a buffer has only a basic pair of methods to get back the original data and its size, reflecting any narrowing that occurred.

void *vptr = mbuf.data();
size_t size = mbuf.size();

And, that's about it.

Almost always we create the default context:

zmq::context_t ctx;

A move constructor is also available. After construction, various context options may be set and queried and cppzmq provides ctx.set() and ctx.get() to do this. I have yet to use them but we may check zmq_ctx_set(3) for anything interesting.

We can not copy a context but we can move. Typically, we will construct it, keep it alive as long as we need the sockets that we have created with it and if we need it in some other code context we may pass it by reference. Here is a contrived example.

void run_app(zmq::context_t& ctx);

void main()
{
    zmq::context_t ctx;
    run_app(ctx);
}

In most applications, we will let the context_t destruct and that will tear down our sockets. We may also tear down even sooner:

// Cease any blocking operations in progress.
ctx.shutdown();

// Do a shutdown, if needed and destroy the context.
ctx.close();

To be useful, sockets must be created with a context (which we covered in section Context) and a socket type identifier. In libzmq the type is identified with an integer, usually as provided by a CPP macro like ZMQ_PUB. In cppzmq (in C++11) we may still use these macros or bare integers or we may use an enum class socket_type.

using namespace zmq;
context_t ctx;
socket_t pub(ctx, ZMQ_PUB);
socket_t sub(ctx, socket_type::sub);

A default socket may be constructed and later swap guts with another socket through a move assignment. A socket may also be move-constructed. But we can not copy a socket.

When a socket destructs or if its socket_t::close() method is explicitly called, the underlying libzmq socket will be destroyed via zmq_close(3).

ZeroMQ sockets link up with each other via transport addresses (see eg zmq_tcp(7), zmq_ipc(7), zmq_inproc(7)). One or more sockets may "bind" to an address and one or more may "connect". At least one of each type of link must be made for messages to be exchanged. Other rules apply. For example, only certain socket types may intercommunicate and some socket types do not work with some transports. These rules are all general to libzmq (not just cppzmq). It's up to us to write applications that follow these rules while cppzmq provides us very simple ways to form the links.

const std::string addr = "tcp://127.0.0.1:5678";
sock1.bind(addr);
sock2.connect(addr);

Note the lack of return values. A zmq::error_t will be thrown if something goes wrong. The linkage may be kept for the entire life cycle of the sockets but some applications may want to explicitly undo these actions:

sock1.unbind(addr);
sock2.disconnect(addr);

We have several ways to send with cppzmq and several more which are deprecated. Here, let's focus on the preferred methods. The choice is to pass a message_t by lvalue reference or by rvalue reference or to pass a const_buffer by value In all cases we must also provide something called "send flags" which we will cover in a little bit

zmq::message_t msg = ...;

// Pass by reference
auto res = sock.send(msg, zmq::send_flags::none);

// Pass by move
auto res = sock.send(std::move(msg), zmq::send_flags::none);

// Pass by buffer
auto res = sock.send(zmq::str_buffer("hello world"), zmq::send_flags::none);

The first two call zmq_msg_send(3) and the last calls msg_send(3) so see those man pages for any nitty gritty details you may want.

The opposite to sending to a socket is receiving from a socket. We may do this with recv(). The misspelling of this verb has a long history in network programming and ZeroMQ and cppzmq cheerfully continues it. Like with send(), we have one recv() method for a message one for a buffer and for both we give recv flags.

// Fill a message passed by reference
auto res = sock.recv(msg, zmq::recv_flags::none);

// Fill a suitably pre-sized mutable_buffer
auto res = sock.recv(buf, zmq::recv_flags::none);

These two examples correspond to the low-level zmq_msg_recv(3) and zmq_recv(3) respectively. When using the buffer to receive the data, take care that it has sufficient capacity to hold the expected message. The care you must take pays back in avoiding a copy.

Usually we develop an application that makes some ZeroMQ sockets, do our bind()/connect() and some send()/recv() and that's all we need for a lot of fun. There's not much detail to worry about. Sometimes we have a tough problem that needs something special and there's very likely a way that past ZeroMQ community members have found a solution. We then just need to find and set the right socket options.

The rather long list of possible options (85 counted today) are given in zmq_setsockopt(3). There they have names like ZMQ_IMMEDIATE or ZMQ_ROUTING_ID which are CPP macros expanding to some integer. For each, cppzmq creates a type tag to use with friendly get()/set() methods on socket_t.

sock.set(zmq::sockopt::immediate, false);
sock.set(zmq::sockopt::routing_id, "me");
auto rid = sock.get(zmq::sockopt::routing_id);

We will still need to carefully read zmq_setsockopt to discover and understand what options may help us, but then applying or querying them with a cppzmq socket is simple.

One special socket option lets us set socket properties. These can be retrieved from messages that pass through the socket as described above in Message metadata.

sock.set(zmq::sockopt::metadata, zmq::str_buffer("X-color:purple");
auto res = sock.recv(msg, zmq::recv_flags::none);
std::string val = msg.gets("X-color");
assert(val == "purple");

Sockets in cppzmq can not be copied (they can be moved) while it is usual that we want various parts of our application code to share access to the same socket. We can pass around an lvalue reference to our socket but that is not always possible or convenient.

To help with this, we are given a zmq::socket_ref which refers to but does not own a socket. With a socket ref our code can do almost everything it would do with a full socket object and it can know if the underlying socket has been closed.

zmq::socket_t sock(ctx, zmq::socket_type::pub);
zmq::socket_ref sref = sock;

// pass ref by value (copy)
do_some_socket_stuff(sref);

// Check if our socket is still there
if (sref == nullptr) {
    respond_to_closure();
}

// Do stuff with collections of socket refs
std::unordered_set<zmq::socket_ref> bag_of_socks;
bag_of_socks.insert(sref);
for (auto s : bag_of_socks) {
    auto res = s.send(bcast, zmq::send_flags::none);
}

Our nice cppzmq socket is a facade over an opaque socket object from libzmq. Rarely do we care about that but there may be cases where we do. For example, if we want to directly call some libzmq function or maybe inter-operate with CZMQ code, we can do so by getting the underlying libzmq socket.

zmq::socket_t sock = ...;
void* libzmq_socket = sock.handle();
// use libzmq to bind.  We would probably never do this as we'd prefer
// to use sock.bind(), but it shows the possibility.
int port = zmq_bind(libzmq_socket, "tcp://*:*");

// Use a CZMQ socket.  We might do something this if we use nice CZMQ
// CLASS like Zyre!
zsock_t czmq_socket = zsock_new_pub("tcp://*:*");
zmq::socket_ref sock(zmq::from_handle, zsock_resolve(czmq_socket));

Most of the time, let's ignore we have a "handle" on the libzmq socket and rest assured that if we ever need it, it will be there.

In the following example, we construct two poller_t instances, one to poll on input and one to poll on output add some sockets and their associated events (pollin/pollout). We then use the pollers to wait up to a timeout. Depending on the return, our application reacts.

// We could combine the pollers but here keep input and output polling
// separate.
zmq::poller_t<> in_poller, out_poller;
// Our application has two input sockets and one output.
in_poller.add(input_socket1, zmq::event_flags::pollout);
in_poller.add(input_socket2, zmq::event_flags::pollout);
out_poller.add(output_socket, zmq::event_flags::pollout);

const std::chrono::milliseconds timeout{100};
std::vector<zio::poller_event<>> in_events(2);
std::vector<zio::poller_event<>> out_events(1);
while (true) {
    const auto nin = in_poller.wait_all(in_events, timeout);
    if (!nin) {
        std::cout << "input timeout, try again" << std::endl;
        continue;
    }
    for (int ind=0; ind<nin; ++ind) {
        zmq::message_t msg;
        auto rres = in_events[ind].socket.recv(msg, zmq::recv_flags::none);

        const auto nout = out_poller.wait_all(out_events, timeout);
        if (!nout) {
            std::cout << "output timeout, freakout" << std::endl;
            abort();
        }
        auto sres = out_events[0].socket.send(msg, zmq::send_flags::none);
    }
} 

So far this tour has taken us through zmq.hpp and here, right in the middle of this section, we move to zmq_addon.hpp. As noted above, this switch of files feels somewhat arbitrary to me. Parts of zmq_addon.hpp are more "core" for my own use while parts of zmq.hpp are not so "core" to me. So, let us not worry that we now switch our focus to a header that on the surface may sound somehow secondary. It's in fact full of more goodies.

The active_poller_t is created and filled with sockets in a similar manner as poller_t. In fact it uses a poller_t under the hood. However, its add() method takes a third argument which is a std::function<void(event_flags)> and that function will be called when that event is seen.

This allows our application the option of structuring its response to events differently. Instead of complexity held inside the loop, it can be placed in functions or lambdas.

void chirp(zmq::event_flags ef);
zmq::active_poller_t ap;
ap.add(sock1, zmq::event_flags::pollin, [](zmq::event_flags ef) {
  std::cout << "sock1 got " << ef << std::endl;
});
ap.add(sock2, zmq::event_flags::pollin, chirp);

while (true) {
    auto n = ap.wait(timeout);
    if (n) {
        std::cout << "got " << n << " sockets hit\n";
    }
    else {
        std::cout << "timeout" << std::endl;
    }
}

ZeroMQ has the concept of "more". As we send a single (single-part) message we also tell the socket, "you know what, next I'm gonna send you another message and I want you to wait until I do that and then you can transmit both of them together". This is done with cppzmq like:

zmq::message_t msg1, msg2;
auto res = sock.send(msg1, zmq::send_flags::sndmore);
res = sock.send(msg2, zmq::send_flags::none);

That's "more" and of course you can have more "more" messages than just the first one. The chain can continue and, as it does, ZeroMQ is collecting these messages behind the scenes, waiting for that final none. When none comes, ZeroMQ races to do its magic to assure every message in the chain is delivered to all linked sockets or none of them are.

On the other end of the transport, ZeroMQ receives the data from these multiple "more" messages and presents it to the application through multiple calls to recv(). We must write our application to expect this chain of separate-but-together messages and cppzmq provides the help we need.

zmq::message_t msg1, msg2;
auto res = sock.recv(msg1, zmq::recv_flags::none);
if (!msg1.more()) {
    std::cout << "Oh dear, sheeps are going astray\n";
}
else {
    auto res = sock.recv(msg2, zmq::recv_flags::none);
}

Enter multipart_t. I hesitate to call this a "message" because it's really multiple messages which are aggregated into a collection (called multipart_t). Behind the scenes a multipart_t::send() and ::recv() are simply handling multiple single-part message_t objects with "more", as described above. Later we will describe how this can be done without even forming a multipart_t by using code which is to send/recv sort of like what buffers are to message_t.

But first multipart_t, which we should think about as if it were an STL collection (and in fact it uses a std::deque<message_t> under the covers). We can construct them in many ways.

multipart_t empty;
multipart_t first_from_recv(sock);
std::vector<int> data(1024,0);
multipart_t first_from_data(data.data(), data.size()*sizeof(int));
multipart_t first_from_string(std::string("hello world"));

As the variable names are meant to imply, these latter constructors fill in an initial element of the multipart_t container. We can then add more message_t elements:

message_t msg;
multipart_t mp, mp2;

// Prefix and postfix concatenation. 
mp.prepend(mp2);
mp.append(mp2);

// push a part in various forms to front, also "add" variants to back.
mp.pushmem(data, size);
mp.pushstr("hello again");
mp.pushtyp(myobject);           // effectively a memcpy
mp.push(msg);
mp.push_back(msg);              // same as push, but STL spelling

We can also remove a message (a part):

// from front
message_t msg = mp.pop();
std::string s = mp.popstr();
MyObject mo = mp.pop<MyObject>();

// from back
message_t msg = mp.remove();

// all
mp.clear();

We can query and iterate on the multipart_t in many ways:

message_t& first = mp.front();
message_t& last = mp.back();
message_t* nth = mp.peek(42);
std::count << mp.size() << " parts:\n";
for (auto& msg : mp) {
    std::cout << msg.size() << " bytes:\n" << std::endl;
    std::cout << msg.str() << std::endl;
}
if (mp.empty()) {
    std::cout << "no parts\n";
}

Finally, getting back to multiple single-part message transmission we may do:

// Clear and receive one or MORE message_t from a socket
mp.recv(sock);

// Send the one or MORE message_t parts to a socket
mp.send(sock);

This happily hides all the "more" business from our delicate eyes.

We just saw how cppzmq can transmit a sequence of multiple, single-part messages in the ZeroMQ "more" way and how the multipart_t can be used to bundle up the "more"'ing inside the application. CZMQ introduced another way to handle multiple, single-part messages and that is to encode them into a single, single-part message inside the application.