Use several transports

One device can use the shmem transport for some channels and the zeromq transport for others. For example, it uses shmem to a process on the same host and zeromq to another host. This guide shows how to configure the transport of each channel and how to make messages without a conversion. It also shows how to receive from channels of different transports.

The device makes one transport factory for each transport that its channels use. For the rules, see How a device selects the transport.

Configure the channels

1. Select the default transport of the device with --transport. The value is zeromq (default) or shmem. All channels without a transport property use it.

2. Add the channel property transport to each channel that uses the other transport. In --channel-config, add transport=zeromq or transport=shmem.

3. Give the peer channel in the other device the same transport.

4. For each shmem channel, run both devices on the same host with the same --session. Do not use the socket types pub and sub.

The multiple-transports example starts sampler1 with shmem as the default transport. Its channel ack is a sub socket, thus it uses zeromq:

examples/multiple-transports/fairmq-start-ex-multiple-transports.sh.in

SAMPLER1="fairmq-ex-multiple-transports-sampler1"
SAMPLER1+=" --id sampler1"
SAMPLER1+=" --severity debug"
SAMPLER1+=" --transport shmem"
SAMPLER1+=" --channel-config name=data1,type=push,method=bind,address=tcp://127.0.0.1:5555"
SAMPLER1+="                  name=ack,type=sub,method=bind,address=tcp://127.0.0.1:5557,transport=zeromq"

examples/multiple-transports/fairmq-start-ex-multiple-transports.sh.in:5–10

The example has no JSON file. In a JSON file for --mq-config, set transport for the channel or for one socket of the channel. A socket without transport uses the value of its channel. This file gives sampler1 the same channels:

{
    "fairMQOptions": {
        "devices": [
            {
                "id": "sampler1",
                "channels": [
                    {
                        "name": "data1",
                        "sockets": [
                            { "type": "push", "method": "bind", "address": "tcp://127.0.0.1:5555" }
                        ]
                    },
                    {
                        "name": "ack",
                        "transport": "zeromq",
                        "sockets": [
                            { "type": "sub", "method": "bind", "address": "tcp://127.0.0.1:5557" }
                        ]
                    }
                ]
            }
        ]
    }
}

Give all sub-channels of one channel the same transport. The pollers and the data callbacks of the device use the transport of the sub-channel 0 for all sub-channels.

Make messages for the right transport

NewMessage() uses the default transport of the device. To make a message with the transport of a channel, use NewMessageFor(channel, index, args…). It takes the same arguments as NewMessage() after the channel name and the sub-channel index. sampler1 makes its data messages with the transport of data1:

examples/multiple-transports/sampler1.cxx

bool ConditionalRun() override
{
    // Creates a message using the transport of channel data1
    fair::mq::MessagePtr msg(NewMessageFor("data1", 0, 1000000));

    // in case of error or transfer interruption, return false to go to IDLE state
    // successfull transfer will return number of bytes transfered (can be 0 if sending an empty message).
    if (Send(msg, "data1") < 0) {
        return false;
    }

    if (fMaxIterations > 0 && ++fNumIterations >= fMaxIterations) {
        LOG(info) << "Configured maximum number of iterations reached. Leaving RUNNING state.";
        return false;
    }

    return true;
}

examples/multiple-transports/sampler1.cxx:28–45

The device has more functions of this kind: NewSimpleMessageFor(), NewStaticMessageFor() and NewUnmanagedRegionFor(). If you have a reference to the channel, call NewMessage() on the channel. AddTransport(Transport::SHM) returns the factory of a transport, and makes it if the device has none.

Send a message of another transport

You can send a message on a channel of another transport. Channel::Send() then converts the message (Channel.h):

Message	Channel	What the channel does
shmem	zeromq	Makes a zeromq message that points to the shared memory buffer. No copy. The shared memory buffer stays allocated until zeromq releases the message.
zeromq	shmem	Copies the data into the shared memory segment, and destroys the zeromq message.
empty	either	Replaces the message with an empty message of the channel transport.

After the call, your MessagePtr points to the new message. Receive() replaces a message object of another transport with an empty message of the channel transport. The sender of the readout example receives shmem messages and sends them on its zeromq channel sr:

examples/readout/sender.cxx

bool HandleData(fair::mq::MessagePtr& msg, int /*index*/)
{
    if (Send(msg, "sr") < 0) {
        return false;
    }

    return true;
}

examples/readout/sender.cxx:24–31

The copy from zeromq to shmem costs time and memory for large messages. Use NewMessageFor() to prevent it.

Receive from channels of different transports

If you register data callbacks with OnData() on channels of different transports, the device polls each transport in its own thread. The callbacks run in these threads. A mutex lets only one callback run at a time (Device.cxx). The sink of the example registers callbacks for data1 and data2:

examples/multiple-transports/sink.cxx

Sink()
{
    // register a handler for data arriving on "data" channel
    OnData("data1", &Sink::HandleData1);
    OnData("data2", &Sink::HandleData2);
}

examples/multiple-transports/sink.cxx:18–23

Exceptions in the threads of the transports

In this mode, an exception in a poll thread, for example in a data callback, does not move the device to the state ERROR. The device logs fair::mq::Device::PollForTransport() failed: <error>, going to ERROR state. and throws again (Device.cxx). The exception leaves the thread, and C++ calls std::terminate(). Catch the exceptions in your callbacks.

The start script gives data1 and data2 of the sink the same transport, shmem. The test script of the example gives data2 the transport zeromq. Then the sink polls shmem and zeromq in two threads:

examples/multiple-transports/test-ex-multiple-transports.sh.in

SINK+=" --transport shmem"
SINK+=" --channel-config name=$chan1,type=pull,method=connect,address=ipc://$chan1Addr"
SINK+="                  name=$chan2,type=pull,method=connect,address=ipc://$chan2Addr,transport=zeromq"
SINK+="                  name=$ackChan,type=pub,method=connect,address=ipc://$ackChanAddr,transport=zeromq"

examples/multiple-transports/test-ex-multiple-transports.sh.in:25–28

One poller can poll channels of one transport only. NewPoller() throws std::runtime_error for channels of different transports. Use one thread for each transport, with its own poller or a blocking Receive(). sampler1 receives the acknowledgements with a blocking Receive() in a second thread:

examples/multiple-transports/sampler1.cxx

void ListenForAcks()
{
    uint64_t numAcks = 0;

    while (!NewStatePending()) {
        fair::mq::MessagePtr ack(NewMessageFor("ack", 0));
        if (Receive(ack, "ack") < 0) {
            break;
        }
        ++numAcks;
    }

    LOG(info) << "Acknowledged " << numAcks << " messages";
}

examples/multiple-transports/sampler1.cxx:52–65

Check the result

1. Start the devices with --severity debug.

2. Find the line Initializing transport for channel <name>[<index>]: <transport> for each sub-channel. The value default means the default transport of the device.

3. Find the line Adding '<transport>' transport once for each transport of the device.

In the code, GetChannel(name, index).GetTransportName() returns the transport of a sub-channel, and msg->GetType() returns the transport of a message.

To run the example, start fairmq-start-ex-multiple-transports.sh in the directory examples/multiple-transports of the build directory. The script opens one xterm window for each device. When sampler1 leaves the state RUNNING, it logs Acknowledged <n> messages.

Problems that can occur

Message in the log	Cause	Fix
Received message is not a valid FairMQ shared memory message	A shmem channel received a message from a zeromq peer.	Give both ends of the channel the same transport.
poller failed: different transports within same poller are not yet supported	NewPoller() got channels of different transports.	Make one poller for each transport.
PUB/SUB socket type is not supported for shared memory transport	A shmem channel has the type pub or sub.	Add transport=zeromq to the channel.
Unknown transport provided: <name>	A channel property transport has an unknown value.	Use default, zeromq or shmem.
invalid transport type provided: <name>	--transport has an unknown value.	Use zeromq or shmem.

A shmem receiver detects a zeromq message only if the message is smaller than the shmem metadata. A larger zeromq message goes into the receiver as shmem metadata. A zeromq receiver gets the shmem metadata as data and reports no error. Thus a wrong transport on one end of a channel does not always cause an error message.

Checklist

• The default transport of the device is the transport of most channels.
• Each other channel has the property transport.
• Both ends of each channel use the same transport.
• All sub-channels of one channel use the same transport.
• Each shmem channel connects devices on one host with the same --session, and is not pub or sub.
• The device makes messages with NewMessageFor(), or with NewMessage() of the channel.
• Each poller polls channels of one transport only.

Related pages

• Transports: the two transports and how a device selects them.
• Messages and multipart: what Send() does with a message.
• Poll several channels: wait for messages on several channels.
• Example: multiple transports: the example of this guide, step by step.
• Design a topology: select the transport of each link.