Channels and sockets

A device sends and receives messages through its channels. A channel has a name and one or more sub-channels, and each sub-channel has one socket. The configuration of a sub-channel sets the type of its socket, how the socket finds its peers, and the transport that moves the messages.

A channel has a name and one or more sub-channels. Each sub-channel owns one socket, with its own type, method and address. The index of the sub-channel is the second argument of GetChannel(), and the third argument of Send() and Receive(). The default index is 0.

Channels and sub-channels

A device keeps its channels in a map from the channel name to a list of sub-channels: GetChannels() returns a fair::mq::ChannelMap, which is a std::unordered_map<std::string, std::vector<Channel>>. Each sub-channel is a fair::mq::Channel with one socket.

A sub-channel has the name <channel>[<index>], for example data-out[1]. The index starts at 0. To select a sub-channel, give the channel name and the index:

• GetChannel("data-out", 1) returns the Channel object.
• Send(msg, "data-out", 1) and Receive(msg, "data-in") send and receive through it. The index is the third argument. Its default is 0.
• GetNumSubChannels("data-out") returns the number of sub-channels.

For a channel or an index that does not exist, GetChannel() logs GetChannel(): 'data-out[2]' does not exist. and throws std::out_of_range (Device.h).

The sampler of the copypush example sends a copy of each message to each sub-channel of its channel data:

examples/copypush/sampler.cxx

bool ConditionalRun() override
{
    // 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
    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);

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

    std::this_thread::sleep_for(std::chrono::seconds(1));

    return true;
}

examples/copypush/sampler.cxx:26–47

Its start script gives data two sub-channels, with one address= key for each sub-channel:

examples/copypush/fairmq-start-ex-copypush.sh.in

SAMPLER="fairmq-ex-copypush-sampler"
SAMPLER+=" --id sampler1"
SAMPLER+=" --channel-config name=data,type=push,method=bind,rateLogging=0,address=tcp://*:5555,address=tcp://*:5556"

examples/copypush/fairmq-start-ex-copypush.sh.in:5–7

A socket can also have several endpoints (see Addresses and endpoints). Then the socket itself distributes the messages to its peers, by the rules of its socket type. Use several sub-channels when your code must select the peer for each message.

Socket types and patterns

FairMQ also accepts dealer, router, xpub and xsub. The shmem transport does not support pub and sub.

The type of a sub-channel is the type of its socket. FairMQ uses the ZeroMQ socket type of the same name (zeromq/Common.h). The shmem transport also uses ZeroMQ sockets of this type, for the metadata of the messages. The column Pattern gives the ZeroMQ behavior of each type. The ZeroMQ documentation of zmq_socket describes it in full.

Type	Peer	Pattern	Example in the repository
push	pull	Pipeline. A push socket with several peers sends each message to one of them.	1-n-1
pull	push	Pipeline. A pull socket receives from all its peers.	1-n-1
pub	sub	Publish and subscribe. Each sub peer gets a copy.	multiple-channels
sub	pub	Publish and subscribe. The zeromq transport subscribes each sub socket to all messages (zeromq/Socket.h).	multiple-channels
req	rep	Request and reply.	req-rep
rep	req	Request and reply.	req-rep
pair	pair	One peer, both directions.	readout
dealer, router, xpub, xsub		The ZeroMQ types of these names.	None

Channel::Validate() accepts these 11 types. For another type, it throws ChannelConfigurationError (Channel.cxx). The tests of the repository cover push/pull, pub/sub, req/rep and pair.

pub and sub with shared memory

The shmem transport does not support pub and sub. It throws SocketError with PUB/SUB socket type is not supported for shared memory transport when the device makes such a socket (shmem/Socket.h).

Behavior of each socket type

FairMQ calls ZeroMQ in a loop (zeromq/Socket.h). Each ZeroMQ call waits up to 100 ms. If ZeroMQ cannot take or give a message in this time (EAGAIN), the loop checks for an interruption and for the timeout of the call. Then it calls ZeroMQ again. Thus Send() returns in one of these ways:

• The number of bytes, when ZeroMQ accepts the message.
• TransferCode::timeout (-2), when the timeout of the call is over. With the timeout 0, Send() does not wait.
• TransferCode::interrupted (-3), when the device gets a transition request.
• TransferCode::error (-1), when ZeroMQ reports another error. FairMQ logs Failed transfer on socket <id>, errno: <n>, reason: <text>. For ETERM (the context terminates), it logs Terminating socket <id> with the severity debug.

Receive() uses the same loop. The shmem transport has the same loop for its metadata message (shmem/Socket.h). What ZeroMQ does with a full queue, a missing peer or a wrong sequence of calls depends on the socket type. The column "ZeroMQ behavior" comes from the ZeroMQ documentation of zmq_socket and zmq_setsockopt, not from FairMQ.

Type	ZeroMQ behavior	Result in FairMQ
push	Sends each message to one peer, in turn (round robin). When the queues of all peers are full, or when it has no peer, the socket blocks and discards nothing. A connecting socket queues messages for its endpoints also before the peer is there.	Send() waits until a queue has space, the timeout is over, or a transition interrupts it. FairMQ does not set ZMQ_IMMEDIATE, so messages can wait in the queue of a peer that never comes.
pull	Receives from all peers in turn (fair queuing).	Receive() waits for a message.
pub	Sends a copy to each connected subscriber. A subscriber that is not connected yet gets nothing. When the queue of a subscriber is full, the socket drops the message for this subscriber and does not block.	Send() returns the number of bytes also for a dropped message, and the counters count it as sent.
sub	Receives only the messages that match its subscriptions. A new subscription reaches the publisher some time after the connect.	The zeromq transport subscribes each sub socket to all messages (zeromq/Socket.h). SetOption() has no option name for ZMQ_SUBSCRIBE (see Set socket options directly), so a sub socket of FairMQ receives all messages.
req	Only the sequence send, receive, send, receive… A reply goes to the last request.	A second Send() without a Receive() returns TransferCode::error. After a receive timeout, the socket still waits for the reply, so the next Send() also fails.
rep	Only the sequence receive, send, receive, send… The reply goes to the peer of the last request.	A Send() before a Receive() returns TransferCode::error.
pair	One peer. When the queue is full, or when no peer is there, the socket blocks.	As for push.
dealer	Sends in turn to its peers and receives from all peers, in any sequence. When the queues of all peers are full, or when it has no peer, the socket blocks. To talk to a rep socket, each message must start with an empty part.	As for push. Use fair::mq::Parts for the empty part.
router	Adds a part with the identity of the peer before each received message. Uses the first part of each sent message to select the peer. Drops a message for an unknown peer or a full queue, and does not block.	FairMQ sets the identity of each socket to <device id>.<sub-channel name>.<type>, for example sampler1.data[0].dealer (zeromq/Socket.h). Receive into fair::mq::Parts. Part 0 is the identity.
xpub, xsub	As pub and sub, but the subscriptions are messages: an xsub socket subscribes when it sends a message that starts with the byte 1.	FairMQ sets no subscription for these types.

The shmem transport uses ZeroMQ sockets of the same types for its metadata, so the same rules apply to the types that it supports. It sends and reads one ZeroMQ frame in each call and expects the metadata in it. Thus it cannot handle the extra parts of dealer and router: the identity part and the empty part. No test of the repository covers dealer, router, xpub and xsub.

The client of the req-rep example receives the reply before it sends the next request:

examples/req-rep/client.cxx

if (Send(req, "data") > 0) {
    if (Receive(rep, "data") >= 0) {
        LOG(info) << "Received reply from server: \"" << std::string(static_cast<char*>(rep->GetData()), rep->GetSize()) << "\"";

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

        std::this_thread::sleep_for(std::chrono::seconds(1));

        return true;
    }
}

examples/req-rep/client.cxx:43–56

A sub that connects late misses the messages that the pub sent before. The test devices of the repository synchronize through a second channel. Each sub waits and sends a ready message. The pub publishes only after it receives the ready messages:

test/helper/devices/TestSub.h

auto Run() -> void override
{
    // A pub socket drops messages for a subscriber until the subscription
    // has reached it. The connect is done, but the subscription can still
    // be in flight: give it time before the ready signal.
    std::this_thread::sleep_for(std::chrono::milliseconds(200));

    auto ready = NewMessage();
    Check(Send(ready, "control") >= 0, "send ready signal");

    auto msg = NewMessage();
    Check(Receive(msg, "data") >= 0, "receive data");

    auto ack = NewMessage();
    Check(Send(ack, "control") >= 0, "send ack");

    Succeed();
}

test/helper/devices/TestSub.h:23–40

Bind and connect

The method of a sub-channel is bind or connect. A binding socket waits for peers on its endpoints. A connecting socket connects to the endpoints of a peer. The method does not depend on the type: a push socket can bind or connect. Usually the device with a fixed place in the topology binds, for example the sampler and the sink in the 1-n-1 example.

The device attaches its sub-channels in two states (Device.cxx):

• In BINDING, the device binds all binding sub-channels, with one attempt. If one fails, the device goes to ERROR with 1 of the binding channels could not initialize.
• In CONNECTING, the device connects all connecting sub-channels. If the address of a sub-channel is not valid or its host name does not resolve, the device tries again every 50 ms. Before each attempt, it reads the address property again. After the time of --init-timeout (default 120 s), it goes to ERROR with could not connect all channels after 120 attempts. The number in this message is the timeout in seconds.

ZeroMQ connects in the background. Thus a connect to a peer that is not there yet does not fail. You can start the devices of a topology in any sequence. GetNumberOfConnectedPeers() returns the number of peers of a sub-channel.

Addresses and endpoints

The address of a sub-channel has one endpoint or several endpoints, separated by ,. An endpoint has the form [modifier]protocol://location:

Protocol	Example	Use
tcp://	tcp://*:5555, tcp://host1:5555	Between hosts, or on one host.
ipc://	ipc:///tmp/fmq_data	Between processes on one host, through a Unix domain socket file.
inproc://	inproc://data	Between sub-channels of one device with the same transport.

Validate() checks the protocol, a : after tcp://, and a location that is not empty after ipc:// and inproc:// (Channel.cxx). It also accepts verbs://, a remainder of the removed ofi transport. The zeromq and shmem transports cannot use it.

In a TCP bind, the host is * for all interfaces, an IPv4 address or a host name. In a TCP connect, the host is an IPv4 address or a host name. The device replaces each host name with its first IPv4 address before it binds or connects (Device.cxx).

A modifier before an endpoint overrides the method for this endpoint (Device.cxx):

• @ binds the endpoint.
• + or > connects the endpoint.

With modifiers, one socket can bind some endpoints and connect others. A sub-channel without a method needs a modifier on each endpoint. The device then attaches it in CONNECTING. In this file of the multipart example, the socket of sampler1 connects to two sinks. sink2 binds with @, although its method is connect:

examples/multipart/ex-multipart-multiple-endpoints.json

{
    "fairMQOptions": {
        "devices": [
            {
                "id": "sampler1",
                "channels": [
                    {
                        "name": "data",
                        "sockets": [
                            {
                                "type": "push",
                                "method": "connect",
                                "address": "tcp://localhost:5555,tcp://localhost:5556",
                                "sndBufSize": 1000,
                                "rcvBufSize": 1000,
                                "rateLogging": 0
                            }
                        ]
                    }
                ]
            },
            {
                "id": "sink1",
                "channels": [
                    {
                        "name": "data",
                        "sockets": [
                            {
                                "type": "pull",
                                "method": "bind",
                                "address": "tcp://*:5555",
                                "sndBufSize": 1000,
                                "rcvBufSize": 1000,
                                "rateLogging": 0
                            }
                        ]
                    }
                ]
            },
            {
                "id": "sink2",
                "channels": [
                    {
                        "name": "data",
                        "sockets": [
                            {
                                "type": "pull",
                                "method": "connect",
                                "address": "@tcp://*:5556",
                                "sndBufSize": 1000,
                                "rcvBufSize": 1000,
                                "rateLogging": 0
                            }
                        ]
                    }
                ]
            }
        ]
    }
}

examples/multipart/ex-multipart-multiple-endpoints.json:1–60

Separators

Separate several endpoints with ,. Validate() splits the address at ;, but the device splits it at , (Device.cxx). An address with ; passes the check and then fails in the transport. In --channel-config, the , separates the keys, so give several endpoints in a JSON file or as a property.

After the attach, the device writes the address that it uses back to the property chans.<name>.<index>.address (Device.cxx). This value has the resolved IPv4 addresses and the port that autobind selected.

Automatic ports

When a TCP bind fails and autoBind is true (the default), the device binds to a random port from portRangeMin to portRangeMax (22000 to 23000). It tries up to 1000 ports (Channel.cxx).

FairMQ uses this for binding sub-channels without an address. Such a sub-channel gets the address tcp://<IP of --network-interface>:1. Without root rights, the bind to port 1 fails, and autobind selects a port. With --network-interface default, the device uses the interface of the default route. Code that knows the topology then reads the address property and gives it to the peers, for example a plugin.

Fixed ports

If the configured port is in use, autobind binds to a random port. The device logs this only with the severity debug, and the peers do not find the device. For fixed ports, set autoBind to false. Channel properties describes the CMake variable that changes the default.

The transport of a sub-channel

Each sub-channel has a transport: zeromq or shmem. The transport property of the sub-channel selects it. With the value default, the sub-channel uses the default transport of the device, from --transport (default zeromq). The device makes one transport object for each transport that its sub-channels use, and the sub-channels share it (Device.cxx).

• Give both ends of a connection the same transport, as the examples do.
• With shmem, all peers must be on the same host and in the same session. Refer to Sessions and segments.
• A poller and the data callbacks of OnData() expect one transport for all sub-channels of a channel.

Transports compares the transports. Use several transports shows a device with channels of both transports.

Queues, timeouts and the rate log

Each sub-channel has more properties. Channel properties gives all details.

Properties	Default	Effect
sndBufSize, rcvBufSize	1000	Maximum number of messages in the send queue and the receive queue.
sndKernelSize, rcvKernelSize	0	Kernel buffer sizes in bytes. 0: the system default.
sndTimeoutMs, rcvTimeoutMs	-1	Timeout of Send() and Receive() without a timeout argument. -1: no timeout.
linger	500	Time in ms to keep unsent messages after the socket closes. See Linger.
rateLogging	1	Interval of the rate log in seconds. 0: no rate log.

In RUNNING, the device logs the rate of each sub-channel with a rateLogging above 0, with the severity info:

data[0]: in: <messages/s> (<MB/s> MB) out: <messages/s> (<MB/s> MB)

Socket counters

Each socket counts its transfers. Channel gives the counters of its socket, and Socket has the same four functions:

Function	Counts
GetMessagesTx()	Messages that Send() gave to ZeroMQ.
GetBytesTx()	Bytes of these messages.
GetMessagesRx()	Messages that Receive() got.
GetBytesRx()	Bytes of these messages.

The rules are in the send and receive functions of the transports (zeromq/Socket.h, shmem/Socket.h):

• The counters start at 0 when the transport makes the socket. In a device, this occurs in BINDING or CONNECTING. After RESET DEVICE, the next initialization makes new sockets, which start at 0 again.
• Only a successful transfer counts. A timeout, an interruption or an error does not change the counters.
• A multipart message counts as one message. Its bytes are the sum of the sizes of its parts.
• The bytes are the sizes of the messages. The shmem transport does not count its metadata message.
• A pub socket counts a message also when ZeroMQ drops it for a subscriber. See Behavior of each socket type.

The counters are unsigned long values that only increase. Another thread can read them during the transfers. To log other values, read the counters in your own code, for example GetChannel("data", 0).GetMessagesTx().

The rate log reads the counters in its own thread. The thread wakes up every second. At each interval, it reads the four counters of each sub-channel and divides the differences by the time since its last wake-up (Device.cxx). Thus with a rateLogging above 1, the logged rates are too high by the factor rateLogging (a defect of the code).

Set socket options directly tells how to set ZeroMQ options of a socket with SetOption().

Related pages

• Design a topology: select the patterns, the binding devices and the addresses.
• Configure channels: give the channels with --channel-config, a JSON file or code.
• Channel properties: all properties, with their defaults and allowed values.
• Device states: the states BINDING and CONNECTING.
• Messages and multipart: what goes through a channel.
• Poll several channels: receive from several sub-channels in one loop.