Use std::pmr memory resources

Each transport factory has a memory resource of the C++17 type std::pmr::memory_resource. A container that allocates from this resource keeps its data in a message buffer of the transport. fair::mq::getMessage() then makes a message from the container without a copy, and you send it as a normal message.

The names in FairMQ

MemoryResources.h declares namespace pmr = std::pmr; in fair::mq. Thus fair::mq::pmr::polymorphic_allocator<T> is std::pmr::polymorphic_allocator<T>, and fair::mq::pmr::vector<T> is std::pmr::vector<T>. Before v1.10.0, fair::mq::pmr was boost::container::pmr. Code that uses boost::container::pmr types with the FairMQ resources does not compile with v1.10.0 and later. Use the std::pmr types.

Name	Header	What it is
fair::mq::MemoryResource	MemoryResources.h	Interface. A pmr::memory_resource that can give the message of a buffer.
fair::mq::ChannelResource	MemoryResources.h	The memory resource of a transport factory. Each allocation is one message of this transport.
FairMQMemoryResource	MemoryResources.h	Deprecated name of fair::mq::MemoryResource.
fair::mq::BytePmrAllocator	MemoryResourceTools.h	pmr::polymorphic_allocator<fair::mq::byte>, where fair::mq::byte is unsigned char.
fair::mq::getMessage()	MemoryResourceTools.h	Makes a message from a container.

TransportFactory.h includes MemoryResources.h. For getMessage() and BytePmrAllocator, include <fairmq/MemoryResourceTools.h>. For the deprecated names, see Move from the FairMQ* names.

The memory resource of a transport

TransportFactory::GetMemoryResource() returns a pointer to the ChannelResource of the factory. Each factory has its own resource, and each call returns the same pointer. The test MemoryResources.transportAllocatorMap checks this:

test/memory_resources/_memory_resources.cxx

auto allocZMQ = factoryZMQ->GetMemoryResource();
auto allocSHM = factorySHM->GetMemoryResource();

EXPECT_NE(allocZMQ, nullptr);
EXPECT_NE(allocSHM, allocZMQ);
auto _tmp = factoryZMQ->GetMemoryResource();
EXPECT_EQ(_tmp, allocZMQ);

test/memory_resources/_memory_resources.cxx:91–97

The conversion operator of the factory gives the same pointer. Thus a factory reference can replace the pointer, for example in getMessage(std::move(v), *factory).

In a device, Transport()->GetMemoryResource() returns the resource of the default transport. For the transport of one channel, use GetChannel("data").Transport()->GetMemoryResource().

The ChannelResource gives each allocation to the transport (MemoryResources.cxx):

• An allocation of n bytes makes a message of n bytes with the requested alignment. The resource keeps the message, with the address of its data as the key.
• A deallocation destroys the message of this address.
• Two resources are equal only if they are the same object.
• getNumberOfMessages() returns the number of messages that the resource keeps.

Send a container without a copy

1. Include <fairmq/MemoryResourceTools.h>.

2. Get the memory resource of the transport of the channel that sends the data. With the resource of another transport, Send() converts the message. See Send a message of another transport.

3. Make a container with a pmr allocator on this resource, for example a fair::mq::pmr::vector.

4. Reserve the full size with reserve() before you add the elements.

5. Add the elements.

6. Call getMessage(std::move(container)). Do not use the container after this call.

7. Send the message.

The test PmrMessage.FromContainerWithoutCopy does these steps for each transport. Then it sends the message on a channel and receives it:

test/memory_resources/_memory_resources.cxx

TEST_P(PmrMessage, FromContainerWithoutCopy)
{
    ProgOptions config;
    test::Configure(config, tools::Uuid(), GetParam().expandedShmMetadata);
    auto factory = TransportFactory::CreateTransportFactory(GetParam().transport, tools::Uuid(), &config);

    std::vector<int, polymorphic_allocator<int>> v(polymorphic_allocator<int>{factory->GetMemoryResource()});
    v.reserve(4);
    v.push_back(7);
    v.push_back(8);
    v.push_back(9);
    void const* data = v.data();

    auto message = getMessage(std::move(v));
    ASSERT_NE(message, nullptr);
    EXPECT_EQ(message->GetData(), data);
    EXPECT_EQ(message->GetSize(), 3 * sizeof(int));
    EXPECT_EQ(factory->GetMemoryResource()->getNumberOfMessages(), 0U);

    Channel push("push", "push", factory);
    Channel pull("pull", "pull", factory);
    test::IpcAddress const address("pmr_message");
    ASSERT_TRUE(pull.Bind(address));
    ASSERT_TRUE(push.Connect(address));
    ASSERT_EQ(push.Send(message, 5000), static_cast<int64_t>(3 * sizeof(int)));

    auto in(pull.NewMessage());
    ASSERT_EQ(pull.Receive(in, 5000), static_cast<int64_t>(3 * sizeof(int)));
    auto const* received = static_cast<int const*>(in->GetData());
    EXPECT_EQ(received[0], 7);
    EXPECT_EQ(received[1], 8);
    EXPECT_EQ(received[2], 9);
}

test/memory_resources/_memory_resources.cxx:186–218

After getMessage(), the message has the address of the container data. For one exception of the shmem transport, see the unused capacity in Limits. The size of the message is size() multiplied by the size of one element, and the resource keeps no message.

In a device, the same steps look like this:

#include <fairmq/Device.h>
#include <fairmq/MemoryResourceTools.h>

#include <utility>
#include <vector>

struct Sampler : fair::mq::Device
{
    bool ConditionalRun() override
    {
        auto* resource = GetChannel("data").Transport()->GetMemoryResource();

        fair::mq::pmr::vector<float> values(resource);
        values.reserve(1000);
        for (int i = 0; i < 1000; ++i) {
            values.push_back(0.5f * i);
        }

        fair::mq::MessagePtr msg(fair::mq::getMessage(std::move(values)));
        return Send(msg, "data") >= 0;
    }
};

What getMessage() does

getMessage(container, target) moves the container into the function, and finds the resource of its allocator (MemoryResourceTools.h). Give the container as an rvalue, for example std::move(v). A call with an lvalue does not compile, because the function uses ContainerT::value_type and ContainerT is then a reference type. The optional target is a fair::mq::MemoryResource*. The result depends on the two resources:

Resource of the container	target	Result
A FairMQ resource	none, or the same resource	No copy. The message of the container buffer, with the size of the elements in the container.
A FairMQ resource	another FairMQ resource	A new message of the target transport with a copy of the data. The container frees its buffer.
Another std::pmr resource	a FairMQ resource	A new message of the target transport with a copy of the data.
Another std::pmr resource	none	Throws std::runtime_error with Neither the container or target resource specified.

If the resource has no message for the container data, getMessage() copies the data into a new message. This occurs for a container without a buffer. Examples are an empty vector without reserve(), and a short std::pmr::string that keeps its characters in the string object.

The test MemoryResources.getMessage shows both results. The first message has the address of the vector data. The second message, with the shmem factory as target, has a new address:

test/memory_resources/_memory_resources.cxx

// test message creation on the same channel it was allocated with
{
    std::vector<TestData, polymorphic_allocator<TestData>> v(polymorphic_allocator<TestData>{allocZMQ});
    v.emplace_back(1);
    v.emplace_back(2);
    v.emplace_back(3);
    void* vectorBeginPtr = &v[0];
    message = getMessage(std::move(v));
    ASSERT_NE(message, nullptr);
    EXPECT_EQ(message->GetData(), vectorBeginPtr);
}
EXPECT_EQ(message->GetSize(), 3 * sizeof(TestData));
messageArray = static_cast<int*>(message->GetData());
EXPECT_EQ(messageArray[0], 1);
EXPECT_EQ(messageArray[1], 2);
EXPECT_EQ(messageArray[2], 3);

// test message creation on a different channel than it was allocated with
{
    std::vector<TestData, polymorphic_allocator<TestData>> v(polymorphic_allocator<TestData>{allocZMQ});
    v.emplace_back(4);
    v.emplace_back(5);
    v.emplace_back(6);
    void* vectorBeginPtr = &v[0];
    message = getMessage(std::move(v), *factorySHM);
    ASSERT_NE(message, nullptr);
    EXPECT_NE(message->GetData(), vectorBeginPtr);
}

EXPECT_EQ(message->GetSize(), 3 * sizeof(TestData));
messageArray = static_cast<int*>(message->GetData());
EXPECT_EQ(messageArray[0], 4);
EXPECT_EQ(messageArray[1], 5);
EXPECT_EQ(messageArray[2], 6);

test/memory_resources/_memory_resources.cxx:146–179

Limits

• Allocator. The container must use a pmr::polymorphic_allocator, because getMessage() calls get_allocator().resource().
• Contiguous data. getMessage() uses data() and size() of the container. The buffer must start at data(), as in std::vector.
• Element type. Use types that you can copy as bytes, for example numbers and structs of numbers. getMessage() destroys the container, and with it the elements. The message keeps their bytes. A receiver in another process also gets only the bytes.
• Growth. Each allocation makes a message. If the vector grows past its capacity, it allocates a new message, copies the elements, and destroys the old message. Use reserve().
• Unused capacity. getMessage() makes the message smaller with SetUsedSize(). The zeromq transport keeps the unused memory until it destroys the message. The shmem transport tries to give the unused memory back to the segment. If this fails, it keeps the memory, or it copies the data into a new buffer if 1000000 bytes or more are unused.
• Threads. ChannelResource has no lock. All channels of one transport in a device share one resource. Allocate and call getMessage() from one thread at a time.
• Lifetime. The resource is a member of the transport factory. Destroy the containers before the factory. A device destroys its factories in the state RESETTING DEVICE.
• Receive. getMessage() is for the sender only. A received message is a normal message: read it with GetData() and GetSize().

Related pages

• Messages and multipart: how to make, send and receive messages.
• Transports: the transport factories that own the memory resources.
• How shared memory works: where the shmem transport keeps the message data.
• Use channels without a device: the channels of the tests in this guide.
• Send your own data types: structs, arrays and serialized objects in messages.