Messages and multipart

A message is the unit of data that a channel sends and receives. The transport allocates the buffer of the message, and the owner of the buffer changes when you send it. A multipart message sends several messages in one transfer.

The message object

fair::mq::Message is an interface. Each transport has its own implementation: the zeromq message holds a zmq_msg_t, the shmem message holds a handle to a buffer in shared memory. You use a message through fair::mq::MessagePtr, which is a std::unique_ptr<Message>. You cannot copy or move a Message object, but you can move the MessagePtr.

Function	What it does
GetData()	Returns a pointer to the buffer, or nullptr if the message has no buffer.
GetSize()	Returns the size of the buffer in bytes.
SetUsedSize(size, alignment)	Makes the buffer smaller. See Change the size.
Rebuild(…)	Releases the buffer and initializes the message again.
Copy(other)	Makes the message use the buffer of other.
GetType()	Returns the transport type, Transport::ZMQ or Transport::SHM.
GetTransport()	Returns a pointer to the TransportFactory that made the message.

Make a message

A device makes messages with NewMessage(args…). It forwards the arguments to CreateMessage() of the default transport of the device. NewMessageFor(channel, index, args…) uses the transport of the sub-channel index of channel. A Channel has a NewMessage(args…) function too.

Arguments	Result
none	An empty message without a buffer. Use it to receive.
Alignment{64}	An empty message to receive into. The zeromq transport aligns the received buffer to 64 bytes.
size	A message with a buffer of size bytes from the transport.
size, Alignment{64}	A buffer of size bytes, aligned to 64 bytes.
data, size, ffn, hint	A message for your buffer. See Your own buffer.
region, data, size, hint	A message for a block of an unmanaged region. See Unmanaged regions.

A std::pmr container can also become a message without a copy: fair::mq::getMessage() makes the message from a container that allocates from the memory resource of the transport. See Use std::pmr memory resources.

fair::mq::Alignment is a struct with one size_t member. The value 0 means "no alignment", which is the default. The shmem transport aligns each buffer to at least alignof(std::max_align_t). It ignores the alignment of an empty message, because the receiver gets the buffer that the sender allocated.

Use messages only while the transports exist

The device makes its transports in the state INITIALIZING DEVICE, after it calls Init(). Do not call NewMessage() in the constructor or in Init(). Call it in InitTask(), Run(), ConditionalRun() or in a data callback. The device destroys its transports in the state RESETTING DEVICE. Destroy the messages that your device class keeps in ResetTask(). In a build with the CMake option FAIRMQ_DEBUG_MODE, the shmem transport checks this before the device calls Reset(). It throws MessageError if messages still exist.

Your own buffer

NewMessage(data, size, ffn, hint) makes a message for a buffer that you allocated. ffn is a deleter of the type fair::mq::FreeFn, void(void* data, void* hint). The transport calls ffn(data, hint) when it does not need the buffer any more. The 1-1 example sends a string this way:

examples/1-1/sampler.cxx

std::string* text = new std::string(fText);

// create message object with a pointer to the data buffer, its size,
// custom deletion function (called when transfer is done),
// and pointer to the object managing the data buffer
fair::mq::MessagePtr msg(NewMessage(
    const_cast<char*>(text->c_str()),
    text->length(),
    [](void* /*data*/, void* object) { delete static_cast<std::string*>(object); },
    text));

examples/1-1/sampler.cxx:28–37

The two transports use the buffer differently:

• The zeromq transport does not copy the buffer. ZeroMQ calls ffn after the transfer. ZeroMQ can make this call from one of its own threads, so ffn must be thread-safe. See Threads in a device.
• The shmem transport copies the buffer into the shared memory segment when it makes the message. Then it calls ffn immediately. If ffn is nullptr, it calls free(data) (shmem/Message.h). For a size of 0, it does not call ffn and does not free the buffer.

Always give a deleter. For a buffer that you release yourself, give TransportFactory::NoCleanup.

Small values and strings

NewSimpleMessage(value) copies value into a new message, and the message owns the copy. For a type T, the message has sizeof(T) bytes. T must be trivially copyable, because the receiver gets the bytes only. For a string literal and a std::string, the message holds the characters without the terminating zero. Limits of NewSimpleMessage() lists the types that do not work.

NewStaticMessage(str) makes a message for the buffer of the std::string str, with the deleter NoCleanup. The zeromq transport does not copy the buffer and does not tell you when the transfer ends. Use it only for a string that does not change and that lives longer than the transport. The transports of a device live until RESETTING DEVICE, so a member of the device is enough. The shmem transport copies the string when it makes the message. The test of the transport factory shows both functions:

test/transport/_factory.cxx

    // NewSimpleMessage() copies the data: the source can change after the call.
    string text("simple");
    auto simple(factory->NewSimpleMessage(text));
    text = "changed";
    ASSERT_EQ(push.Send(simple, 5000), 6);
    EXPECT_EQ(AsString(*receive()), "simple");

    auto literal(factory->NewSimpleMessage("literal"));
    ASSERT_EQ(push.Send(literal, 5000), 7);
    EXPECT_EQ(AsString(*receive()), "literal");

    uint64_t const value = 0x0123456789abcdef;
    auto number(factory->NewSimpleMessage(value));
    ASSERT_EQ(push.Send(number, 5000), static_cast<int64_t>(sizeof(value)));
    auto numberIn = receive();
    ASSERT_EQ(numberIn->GetSize(), sizeof(value));
    uint64_t received = 0;
    memcpy(&received, numberIn->GetData(), sizeof(received));
    EXPECT_EQ(received, value);

    // NewStaticMessage() uses the buffer of the caller, which must outlive the message.
    static string const staticText("static");
    auto staticMsg(factory->NewStaticMessage(staticText));
    EXPECT_EQ(staticMsg->GetSize(), staticText.size());
    if (GetParam().transport == "zeromq") {
        EXPECT_EQ(staticMsg->GetData(), staticText.data()); // zeromq does not copy
    }
    ASSERT_EQ(push.Send(staticMsg, 5000), 6);
    EXPECT_EQ(AsString(*receive()), "static");
}

test/transport/_factory.cxx:116–145

Caution

Use NewStaticMessage() with a std::string only. The template overload NewStaticMessage(const T& data) gives data itself as the buffer pointer and sizeof(T) as the size (TransportFactory.h).

The functions NewSimpleMessageFor() and NewStaticMessageFor() take a channel name and a sub-channel index first, like NewMessageFor().

Change the size

SetUsedSize(size) makes the buffer smaller. Use it when you allocate the maximum size first and know the real size later. It returns true if the size is equal to or smaller than the current size, and false if the size is larger. The message test shows the rules:

test/message/_message.cxx

auto outMsg(push.NewMessage(size));
ASSERT_EQ(outMsg->GetSize(), size);
memcpy(outMsg->GetData(), "ABCDEF", size);
ASSERT_TRUE(outMsg->SetUsedSize(5));
ASSERT_TRUE(outMsg->SetUsedSize(5));
ASSERT_FALSE(outMsg->SetUsedSize(7));
ASSERT_EQ(outMsg->GetSize(), 5);

test/message/_message.cxx:61–67

The zeromq transport keeps the full buffer until it releases the message. The shmem transport shrinks the buffer in the segment where possible, and SetUsedSize(0) releases the buffer.

Rebuild and copy a message

Rebuild() releases the buffer of the message and initializes the message again. It has five overloads with the same arguments as NewMessage(), but without the region overload. Use Rebuild() to use one message object for many transfers.

Copy(other) makes the message use the buffer of other. The current transports share the buffer and do not copy the data, but the interface does not guarantee it. The shmem transport increments a reference count in shared memory, also for messages of other processes. The zeromq transport calls zmq_msg_copy(). Do not change the buffer after Copy(). Both messages must use the same transport. The copypush example sends one buffer on several sub-channels:

examples/copypush/sampler.cxx

fair::mq::MessagePtr msg(NewSimpleMessage(fCounter++));

for (int i = 0; i < fNumDataChannels - 1; ++i) {
    fair::mq::MessagePtr msgCopy(NewMessage());
    msgCopy->Copy(*msg);
    Send(msgCopy, "data", i);
}
Send(msg, "data", fNumDataChannels - 1);

examples/copypush/sampler.cxx:30–37

Send and receive a message

Send(msg, channel, index) takes a reference to the MessagePtr. It returns the number of bytes, or a negative transfer code for a timeout, an interruption or an error.

After a successful Send(), the transport owns the buffer. The MessagePtr still holds a message object, but you must not read or write its data:

• The zeromq transport gives the message to zmq_msg_send(). On success, ZeroMQ empties the message, and GetSize() returns 0.
• The shmem transport keeps the old size and pointer in the message. The buffer belongs to the receiver now.

To send the same data on two channels, make a second message with Copy() before the first Send(). To use the message object again, call Rebuild(). If Send() returns a negative value, the message keeps its buffer, and you can send it again.

Receive(msg, channel, index) puts the received buffer into msg. Receive into an empty message: make a new one with NewMessage(), or call Rebuild() on a used one. The shmem transport does not release a buffer that the message already holds. The received message owns its buffer. When you destroy the message, the transport releases the buffer.

A message of another transport

Channel::Send() compares the transport of each message with the transport of the channel (Channel.h). If they are different, the channel makes a new message of its own transport, and your MessagePtr points to the new message after the call:

• If the channel uses zeromq, the new message points to the buffer of your message. FairMQ does not copy the data. It destroys your message when zeromq releases the buffer.
• If the channel uses shmem, the new message has a copy of the data in shared memory. FairMQ destroys your message immediately.
• An empty message becomes an empty message of the channel transport.

Channel::Receive() replaces a message of another transport with an empty message of the channel transport. To prevent the conversion, make the message with NewMessageFor(). See Use several transports.

Multipart messages

fair::mq::Parts holds a std::vector<MessagePtr>. Each message in it is a part. You can move a Parts object, but you cannot copy it.

Function	What it does
Parts(part1, part2, …)	Makes a Parts object from MessagePtr values.
AddPart(part)	Adds a MessagePtr at the end.
AddPart(part1, part2, …)	Adds several parts at the end.
AddPart(otherParts)	Moves all parts of another Parts object to the end.
operator[](i)	Returns the part i as Message&. It does not check the index.
At(i)	Returns the MessagePtr& of the part i, and throws std::out_of_range for a bad index. You can replace the part through it.
Size(), Empty(), Clear()	Return the number of parts, check for zero parts, remove all parts.
begin(), end(), cbegin(), cend()	Iterate over the MessagePtr elements, for example in a range-based for loop.

AddPart(Message*) and AtRef(i) are deprecated. The old name FairMQParts is an alias of fair::mq::Parts (see Move from the FairMQ* names).

The sampler of the multipart example makes seven parts. It moves three of them from a second Parts object with AddPart(), and one part is empty:

examples/multipart/sampler.cxx

fair::mq::Parts parts;

// NewSimpleMessage creates a copy of the data and takes care of its destruction (after the transfer takes place).
// Should only be used for small data because of the cost of an additional copy
parts.AddPart(NewSimpleMessage(header));
parts.AddPart(NewMessage(1000));

// create more data parts, testing the fair::mq::Parts in-place constructor
fair::mq::Parts auxData{ NewMessage(500), NewMessage(600), NewMessage(700) };
assert(auxData.Size() == 3);
parts.AddPart(std::move(auxData));
assert(auxData.Size() == 0);
assert(parts.Size() == 5);

parts.AddPart(NewMessage());

assert(parts.Size() == 6);

parts.AddPart(NewMessage(100));

assert(parts.Size() == 7);

LOG(info) << "Sending body of size: " << parts.At(1)->GetSize();

Send(parts, "data");

examples/multipart/sampler.cxx:38–62

The sink receives all parts in one data callback:

examples/multipart/sink.cxx

bool HandleData(fair::mq::Parts& parts, int)
{
    LOG(info) << "Received message with " << parts.Size() << " parts";

    if (parts.Size() != 7) {
        throw std::runtime_error("Number of received parts != 7");
    }

    example_multipart::Header header;
    header.stopFlag = (static_cast<example_multipart::Header*>(parts.At(0)->GetData()))->stopFlag;

examples/multipart/sink.cxx:23–32

How a multipart message travels

Send(parts, …) sends all parts in one transfer. The receiver gets all parts, in the same order, in one Receive(parts, …) call. Send() returns the sum of the sizes of the parts. The channel counts the transfer as one message in its statistics.

• The zeromq transport sends one zeromq multipart message, with one frame for each part.
• The shmem transport sends one metadata message with the number of parts and the metadata of each part. The data of the parts stays in shared memory.

Send() and Receive() also accept a std::vector<MessagePtr>. The rules for single messages apply to each part. This test sends three parts, one of them empty, and receives them:

test/parts/_access.cxx

Parts out(push.NewSimpleMessage("ab"), push.NewMessage(), push.NewSimpleMessage("cde"));
ASSERT_EQ(push.Send(out, 5000), 5);

Parts in;
ASSERT_EQ(pull.Receive(in, 5000), 5);
ASSERT_EQ(in.Size(), 3U);
EXPECT_EQ(AsString(in[0]), "ab");
EXPECT_EQ(in[1].GetSize(), 0U);
EXPECT_EQ(AsString(in[2]), "cde");

test/parts/_access.cxx:110–118

Rules for multipart messages

• Do not send an empty Parts object. The zeromq transport returns TransferCode::error, but the shmem transport sends a message with zero parts.
• Receive with the same form that the sender uses: Parts for Parts, MessagePtr for MessagePtr. The shmem transport cannot read a single message into a Parts object.
• Receive() adds the received parts at the end of the Parts object. Give it an empty one.
• A receiver with OnData() must register a callback that takes fair::mq::Parts&.

Related pages

• Channels and sockets: the channels that send and receive the messages.
• Transports: how zeromq and shmem move the data.
• Use several transports: make messages for the right transport.
• Unmanaged regions: messages for blocks of your own shared memory.
• Transfer codes and errors: the return values of Send() and Receive().
• Send your own data types: structs, arrays and serialized objects in messages.
• Use std::pmr memory resources: send a std::pmr container without a copy.