Sending Cap'n Proto via ZeroMQ

[op...@vs.uni-kassel.de]

Hi all,

like some other people before me, I would like to send Cap'n Proto Messages via ZeroMQ. Nevertheless I did not manage to make it work, yet. The exception I receive on the receiver side is:

Exception catched:  Receiver - src/capnp/serialize.c++:43: failed: expected array.size() >= offset; Message ends prematurely in segment table.

Here is the sending code:

// cap'n proto part

::capnp::MallocMessageBuilder msgBuilder;
discovery_msgs::Beacon::Builder beaconMsgBuilder = msgBuilder.initRoot<discovery_msgs::Beacon>();
beaconMsgBuilder.setIp(this->wirelessIpAddress);
beaconMsgBuilder.setPort(6666);
beaconMsgBuilder.setUuid(kj::arrayPtr(this->uuid, sizeof(this->uuid)));

auto byteArray = capnp::messageToFlatArray(msgBuilder).asBytes();

// zmq part

this->ctx = zmq_ctx_new();
this->socket = zmq_socket(ctx, ZMQ_RADIO);
zmq_connect(this->socket, "udp://224.0.0.1:5555");
zmq_msg_t msg;
zmq_msg_init_data(&msg, byteArray.begin(), byteArray.size(), NULL, NULL);
zmq_msg_set_group(&msg, "TestMCGroup");
zmq_msg_send(&msg, this->socket, 0);
zmq_msg_close(&msg);

On the receiving side it is:

this->ctx = zmq_ctx_new();
this->socket = zmq_socket(ctx, ZMQ_DISH);
zmq_bind(socket, "udp://224.0.0.1:5555");
zmq_join(this->socket, "TestMCGroup")

zmq_msg_t msg;
zmq_msg_init(&msg);
int nbytes = zmq_msg_recv(&msg, this->socket, 0);

// Received message must contain an integral number of words.
assert(zmq_msg_size(&msg) % sizeof(capnp::word) == 0);
auto num_words = zmq_msg_size(&msg) / sizeof(capnp::word);

if (reinterpret_cast<uintptr_t>(zmq_msg_data(&msg)) % sizeof(capnp::word) == 0) {
    // message is aligned
    auto wordArray = kj::ArrayPtr<capnp::word const>(reinterpret_cast<capnp::word const *>(zmq_msg_data(&msg)), num_words);
    ::capnp::FlatArrayMessageReader msgReader = ::capnp::FlatArrayMessageReader(wordArray); // <-- Throws Exception:  Receiver - src/capnp/serialize.c++:43: failed: expected array.size() >= offset; Message ends prematurely in segment table.
    auto beacon = msgReader.getRoot<discovery_msgs::Beacon>();
    check(zmq_msg_close(&msg), "zmq_msg_close");
} else {
    // message is not aligned
    std::cerr << "Agent: receive(): Not aligned " << std::endl;
}

In my own words, I simply try to send a message via UDP-Multicast over a zmq RADIO socket to a zmq DISH socket. My experience is that the exact same bytes are received, that are also sended before. The message is also always memory aligned. The only problem is the recreation of the Cap'n Proto message. The constructor of the FlatArrayMessageReader throws an exception as indicated in the code above:
Receiver - src/capnp/serialize.c++:43: failed: expected array.size() >= offset; Message ends prematurely in segment table.

My interpretation would be, that the message is not correctly encoded, or not correctly aligned in memory (maybe my alignment check is wrong).

Does somebody have a clue, what I am doing wrong?

Kind Regards
  Stephan

[Kenton Varda]

Hi Stephan,

For some reason Google Groups thought your message was spam, but I fished it out of the spam bucket. (It looked like you sent three different versions. I accepted this one and deleted the other two.)

I think your problem may be here:

    auto byteArray = capnp::messageToFlatArray(msgBuilder).asBytes();

asBytes() returns an ArrayPtr which points back at the array on which it was called. But in this line, you're calling it on a temporary value (the return value of messageToFlatArray), so you end up with `byteArray` being a dangling pointer (and so its contents will be garbage). You could fix that like this:

    auto wordArray = capnp::messageToFlatArray(msgBuilder);

    auto byteArray = wordArray.asBytes();

Or like this:

    auto byteArray = capnp::messageToFlatArray(msgBuilder).releaseAsBytes();

-Kenton

--
You received this message because you are subscribed to the Google Groups "Cap'n Proto" group.
To unsubscribe from this group and stop receiving emails from it, send an email to capnproto+unsubscribe@googlegroups.com.
Visit this group at https://groups.google.com/group/capnproto.

[Stephan Opfer]

Hi Kenton,

thanks for fishing my message out of the spam filter (was driving crazy to write the message a third time :) ).

Currently I have changed my code a little. I send the wordArray now directly. That works, as long as I keep the capnproto message in memory, until zmq has sent its content. That is, because I try to sent it with zero-copy semantic.

So my situation/problem is like this:

zmq_msg_init_data(&msg, byteArray.begin(), byteArray.size(), NULL, NULL);

This is the crucial call to zeromq for creating a zeromq message with zero-copy effort. Here is the API-Reference for this method.
Short arguments summary: 

1. The zeromq message to be created
2. Pointer to the start of the data or content of the message
3. The size of the data, the pointer is pointing to.
4. Function-Pointer of the form void(void*,void*). This method will be called, when zeromq cleans up the message content. First argument of the function is the pointer to the messages content. The second argument is the pointer you can pass as 5. argument (see next).
   
5. void* that will be the second argument of the function in argument 4.

The question is: How can I keep the message content, until zeromq cleaned it up and how to I recognize that zeromq clean it up.

Passing a shared_ptr as 5th argument does not work, according to the zeromq dev mailing list. It also seems that the function (4th argument) must be static (not sure about that).

Basically my c++ software engineering skills come to an end here. I think, that it must work somehow, but I couldn't find it out, so far. Some snippets would be nice... :)

Greetings,
  Stephan

To unsubscribe from this group and stop receiving emails from it, send an email to capnproto+...@googlegroups.com.

Visit this group at https://groups.google.com/group/capnproto.

-- 
Distributed Systems Research Group
Stephan Opfer  T. +49 561 804-6280  F. +49 561 804-6277
Univ. Kassel,  FB 16,  Wilhelmshöher Allee 73,  D-34121 Kassel 
WWW: http://www.uni-kassel.de/go/vs_stephan-opfer/

[Kenton Varda]

Hi Stephan,

You need to make sure the kj::Array<capnp::word> returned by messageToFlatArray() is not destroyed until you're done with it. You can move it to the heap like:

    kj::Array<capnp::word>* arrayPtr = new kj::Array<capnp::word>(kj::mv(wordArray)):

Then you can define a free function:

    template <typename T>

    void freeArray(void *data, void *hint) {

      delete reinterpret_cast<kj::Array<T>*>(hint);

    }

Then you can pass these to zmq:

    zmq_msg_init_data(&msg, byteArray.begin(), byteArray.size(), &freeArray<capnp::word>, ptr);

This is all standard C/C++ programming here, not specific to Cap'n Proto nor KJ. I recommend reading up on C++11 move semantics and RAII, as KJ and Cap'n Proto use these very heavily, so it'll be hard to understand how to use them if you aren't comfortable with these topics. Of course, since ZMQ is a C interface, you need to abandon RAII a bit in this example, creating a bare pointer and a free function instead...

-Kenton

On Tue, Sep 5, 2017 at 12:40 AM, Stephan Opfer <op...@vs.uni-kassel.de> wrote:

Hi Kenton,

thanks for fishing my message out of the spam filter (was driving crazy to write the message a third time :) ).

Currently I have changed my code a little. I send the wordArray now directly. That works, as long as I keep the capnproto message in memory, until zmq has sent its content. That is, because I try to sent it with zero-copy semantic.

So my situation/problem is like this:

zmq_msg_init_data(&msg, byteArray.begin(), byteArray.size(), NULL, NULL);

This is the crucial call to zeromq for creating a zeromq message with zero-copy effort. Here is the API-Reference for this method.
Short arguments summary: 

1. The zeromq message to be created
2. Pointer to the start of the data or content of the message
3. The size of the data, the pointer is pointing to.
4. Function-Pointer of the form void(void*,void*). This method will be called, when zeromq cleans up the message content. First argument of the function is the pointer to the messages content. The second argument is the pointer you can pass as 5. argument (see next).
   
5. void* that will be the second argument of the function in argument 4.

The question is: How can I keep the message content, until zeromq cleaned it up and how to I recognize that zeromq clean it up.

Passing a shared_ptr as 5th argument does not work, according to the zeromq dev mailing list. It also seems that the function (4th argument) must be static (not sure about that).

Basically my c++ software engineering skills come to an end here. I think, that it must work somehow, but I couldn't find it out, so far. Some snippets would be nice... :)

Greetings,
  Stephan

-Kenton

To unsubscribe from this group and stop receiving emails from it, send an email to capnproto+...@googlegroups.com.

Visit this group at https://groups.google.com/group/capnproto.