Built-in devices

FairMQ builds six ready-made devices as executables. Use them to make test traffic, to measure the throughput of a transport, and to move messages between your own devices without new code.

Executable	Class	What it does	Channels (default names)	--multipart
fairmq-bsampler	fair::mq::BenchmarkSampler	Sends messages of a fixed size, at full speed or at a fixed rate.	output data	false
fairmq-sink	fair::mq::Sink	Receives messages. Discards them or writes them to a file.	input data	false
fairmq-splitter	fair::mq::Splitter	Sends each message to the next sub-channel of the output channel (round robin).	input data-in, output data-out	true
fairmq-merger	fair::mq::Merger	Receives from all sub-channels of the input channel and sends on one output channel.	input data-in, output data-out	true
fairmq-multiplier	fair::mq::Multiplier	Sends each message to all sub-channels of all output channels.	input data-in, outputs from --out-channel (no default)	true
fairmq-proxy	fair::mq::Proxy	Receives on the input channel and sends on the output channel.	input data-in, output data-out	true

The last column shows the default value of the option --multipart.

Find the executables

CMake builds the devices together with the FairMQ library (option BUILD_FAIRMQ, default ON).

• In the build directory, the executables are in <build>/fairmq/, for example build/fairmq/fairmq-sink.
• An installation puts them into the bin directory of the install prefix (CMAKE_INSTALL_BINDIR).
• The CMake package exports each executable as an imported target, for example FairMQ::fairmq-sink.

The sources are in fairmq/devices/. Each device is a class in a header, for example Sink.h, and a short run*.cxx file adds its options. The headers are private: an installation does not include them. To change the behavior of a device, write your own device.

Common behavior

• Each run*.cxx file includes <fairmq/runDevice.h>. So each executable has all device options, for example --id, --transport, --channel-config, --control and --severity.
• The device reads its own options in InitTask().
• Configure the channels with --channel-config or --mq-config. See Configure channels. The channel names must be the names that --in-channel and --out-channel give. If a channel does not exist, the device goes to the state ERROR.
• Each address in a --channel-config value makes one sub-channel. For example, name=data-out,type=push,method=bind,address=tcp://*:5556,address=tcp://*:5557 makes the sub-channels data-out[0] and data-out[1].
• Both controllers move the device to RUNNING. Then the default controller (--control dynamic) waits for keys in the terminal. With --control static, the device stops and the process ends when the device leaves RUNNING on its own.
• Each channel logs its message rate and data rate every second (channel property rateLogging, default 1). See Run a benchmark.

Use the same --multipart value in a chain

The default of --multipart is false for the sampler and the sink, and true for the other devices. Give all devices that exchange messages the same value. The shmem transport sends single messages and multipart messages in two different metadata formats. A receiver with the wrong value cannot read the metadata.

fairmq-bsampler

The class fair::mq::BenchmarkSampler makes traffic for benchmarks. Its Run() loop does these steps until the state changes:

• It makes a message of --msg-size bytes with the transport of the output channel. With --multipart true, it makes a multipart message of --num-parts parts. Each part has --msg-size bytes.
• With --memset true, it writes zeros into each buffer. Otherwise it does not write into the buffers.
• It sends the message on sub-channel 0 of --out-channel. With the default send timeout (sndTimeoutMs -1), the send waits until the socket accepts the message.
• With --msg-rate above 0, a fair::mq::tools::RateLimiter limits the number of messages per second. See RateLimit.h.

Option	Default	Description
--out-channelstring	data	Name of the output channel. The sampler sends on its sub-channel 0.runBenchmarkSampler.cxx
--multipartbool	false	Send multipart messages of --num-parts parts. false: send single messages.runBenchmarkSampler.cxx
--memsetbool	false	Write zeros into each buffer before the send.runBenchmarkSampler.cxx
--num-partssize_t	1	Number of parts of each multipart message. Only with --multipart true.runBenchmarkSampler.cxx
--msg-sizesize_t	1000000	Size of each message in bytes. In a multipart message, the size of each part.runBenchmarkSampler.cxx
--msg-alignmentsize_t	0	Alignment of the buffers in bytes. 0: the default alignment of the transport.runBenchmarkSampler.cxx
--max-iterationsuint64_t	0	Number of messages to send. 0: no limit. See the note below.runBenchmarkSampler.cxx
--msg-ratefloat	0	Maximum number of messages per second. 0: no limit.runBenchmarkSampler.cxx

Number of messages

The sampler counts each successful send. With --max-iterations n, it stops at the first send after the count is n. So it sends n + 1 messages and logs Done n iterations. The sink counts in the same way. A sampler and a sink with the same value stop after the same number of messages.

The sampler logs these lines at the start and at the end of Run():

Starting the benchmark with message size of <bytes> and <n> iterations.
Done <n> iterations in <t>ms.

Send messages of 10000 bytes, 100 messages per second:

fairmq-bsampler --id bsampler1 --msg-size 10000 --msg-rate 100 \
  --channel-config name=data,type=push,method=bind,address=tcp://*:5555

fairmq-sink

The class fair::mq::Sink receives messages on sub-channel 0 of --in-channel in its Run() loop. It discards the messages. With --out-filename, it writes the data of each message, or of each part, to the file. The sink leaves RUNNING on its own in two cases:

• It received the number of messages of --max-iterations (it counts like the sampler).
• The file has --max-file-size bytes or more. The sink checks the size after each message.

Option	Default	Description
--in-channelstring	data	Name of the input channel. The sink receives on its sub-channel 0.runSink.cxx
--out-filenamestring	""	Write the data of the received messages to this file (binary). Empty: discard the messages.runSink.cxx
--max-file-sizeuint64_t	2000000000	Leave RUNNING when the file has this number of bytes. 0: no limit. Only with --out-filename.runSink.cxx
--max-iterationsuint64_t	0	Number of messages to receive. 0: no limit.runSink.cxx
--multipartbool	false	Receive multipart messages. Use the value of the sender.runSink.cxx

The sink logs these lines. The line Closed … comes only with --out-filename.

Starting sink and expecting to receive <n> messages.
Received <n> messages in <t>ms.
Closed '<file>' after writing <bytes> bytes.(<rate> MB/s)
Leaving RUNNING state.

When the sink stops on its own, it first logs Configured maximum number of iterations reached. or Written <bytes> bytes, stopping....

If the sink cannot open or write the file, it throws an exception and goes to the state ERROR.

Write 100 MB of received data to a file:

fairmq-sink --id sink1 --out-filename /tmp/data.bin --max-file-size 100000000 \
  --channel-config name=data,type=pull,method=connect,address=tcp://localhost:5555

fairmq-splitter

The class fair::mq::Splitter registers a data callback for --in-channel with OnData(). The callback sends each message to the next sub-channel of --out-channel: 0, 1, …, n − 1, then 0 again. It does not check the result of the send.

fairmq/devices/Splitter.h

template<typename T>
bool HandleData(T& payload, int)
{
    Send(payload, fOutChannelName, fDirection);

    if (++fDirection >= fNumOutputs) {
        fDirection = 0;
    }

    return true;
}

fairmq/devices/Splitter.h:43–53

If --in-channel has more than one sub-channel, the device receives from all of them.

Option	Default	Description
--in-channelstring	data-in	Name of the input channel.runSplitter.cxx
--out-channelstring	data-out	Name of the output channel. The splitter sends to its sub-channels in turn.runSplitter.cxx
--multipartbool	true	Receive and send multipart messages. false: single messages.runSplitter.cxx

Send the messages from port 5555 alternately to port 5556 and port 5557:

fairmq-splitter --id splitter1 \
  --channel-config name=data-in,type=pull,method=bind,address=tcp://*:5555 \
                   name=data-out,type=push,method=bind,address=tcp://*:5556,address=tcp://*:5557

fairmq-merger

The class fair::mq::Merger makes a poller for all sub-channels of --in-channel. In its Run() loop, it waits up to 100 ms for data. Then it receives from each sub-channel that has data and sends the message on sub-channel 0 of --out-channel.

Option	Default	Description
--in-channelstring	data-in	Name of the input channel. The merger receives from all its sub-channels.runMerger.cxx
--out-channelstring	data-out	Name of the output channel. The merger sends on its sub-channel 0.runMerger.cxx
--multipartbool	true	Receive and send multipart messages. false: single messages.runMerger.cxx

Receive on port 5556 and port 5557, and send everything to port 5558:

fairmq-merger --id merger1 \
  --channel-config name=data-in,type=pull,method=bind,address=tcp://*:5556,address=tcp://*:5557 \
                   name=data-out,type=push,method=connect,address=tcp://localhost:5558

fairmq-multiplier

The class fair::mq::Multiplier reads a list of channel names from --out-channel. This option has no default. Without it, InitTask() throws an exception and the device goes to the state ERROR.

The data callback of --in-channel sends each message to all sub-channels of all output channels. The last sub-channel of the last channel gets the original message. Each other sub-channel gets a copy that the device makes with Copy(). The copy shares the buffer with the original, so the device does not copy the data.

Option	Default	Description
--in-channelstring	data-in	Name of the input channel.runMultiplier.cxx
--out-channelvector<std::string>, several values	—	Names of the output channels. Give one or more names. Required.runMultiplier.cxx
--multipartbool	true	Receive and send multipart messages. false: single messages.runMultiplier.cxx

Send each message to two output channels:

fairmq-multiplier --id multiplier1 --out-channel data-out1 data-out2 \
  --channel-config name=data-in,type=pull,method=connect,address=tcp://localhost:5555 \
                   name=data-out1,type=push,method=bind,address=tcp://*:5556 \
                   name=data-out2,type=push,method=bind,address=tcp://*:5557

fairmq-proxy

The class fair::mq::Proxy receives on sub-channel 0 of --in-channel and sends on sub-channel 0 of --out-channel. It leaves RUNNING when the state changes or when a transfer fails. The two channels can have different socket types and different transports. See Use several transports.

Option	Default	Description
--in-channelstring	data-in	Name of the input channel. The proxy receives on its sub-channel 0.runProxy.cxx
--out-channelstring	data-out	Name of the output channel. The proxy sends on its sub-channel 0.runProxy.cxx
--multipartbool	true	Receive and send multipart messages. false: single messages.runProxy.cxx

Receive on a pull channel and publish the messages on a pub channel:

fairmq-proxy --id proxy1 \
  --channel-config name=data-in,type=pull,method=bind,address=tcp://*:5555 \
                   name=data-out,type=pub,method=bind,address=tcp://*:5556

The script startMQBenchmark.sh

CMake makes the script <build>/fairmq/startMQBenchmark.sh from fairmq/devices/startMQBenchmark.sh.in. An installation does not include it. The script starts fairmq-bsampler and fairmq-sink from the build directory, each in its own xterm window. So it needs xterm and a display.

The script takes six positional parameters. A parameter that does not match the accepted characters keeps its default.

Position	Parameter	Default	Accepted value
1	Message size in bytes (--msg-size)	1000000	digits
2	Number of iterations (--max-iterations)	0 (no limit)	digits
3	Transport (--transport)	zeromq	lowercase letters: zeromq or shmem
4	Affinity	false	true: run the sampler with taskset -c 0 and the sink with taskset -c 1
5	Multipart (--multipart)	false	lowercase letters: true or false
6	Number of parts (--num-parts)	1	digits

The usage line that the script prints names only the first four parameters. The script gives these options to the two devices:

fairmq/devices/startMQBenchmark.sh.in

SAMPLER="fairmq-bsampler"
SAMPLER+=" --id bsampler1"
SAMPLER+=" --shm-monitor true"
#SAMPLER+=" --io-threads 2"
#SAMPLER+=" --control static"
SAMPLER+=" --transport $transport"
SAMPLER+=" --severity debug"
SAMPLER+=" --msg-size $msgSize"
SAMPLER+=" --multipart $multipart"
SAMPLER+=" --num-parts $numParts"
SAMPLER+=" --shm-throw-bad-alloc false"
# SAMPLER+=" --shm-metadata-msg-size 1024"
# SAMPLER+=" --msg-rate 1000"
SAMPLER+=" --max-iterations $maxIterations"
SAMPLER+=" --channel-config name=data,type=pair,method=bind,address=tcp://127.0.0.1:5555"
xterm -geometry 90x50+0+0 -hold -e $affinitySamp @CMAKE_CURRENT_BINARY_DIR@/$SAMPLER &
echo ""
echo "started: xterm -geometry 90x50+0+0 -hold -e $affinitySamp @CMAKE_CURRENT_BINARY_DIR@/$SAMPLER"
echo "pid: $!"

SINK="fairmq-sink"
SINK+=" --id sink1"
SINK+=" --shm-monitor true"
#SINK+=" --io-threads 2"
#SINK+=" --control static"
SINK+=" --transport $transport"
SINK+=" --severity debug"
SINK+=" --multipart $multipart"
SINK+=" --max-iterations $maxIterations"
SINK+=" --channel-config name=data,type=pair,method=connect,address=tcp://127.0.0.1:5555"
xterm -geometry 90x50+550+0 -hold -e $affinitySink @CMAKE_CURRENT_BINARY_DIR@/$SINK &

fairmq/devices/startMQBenchmark.sh.in:64–94

The two devices connect with a pair channel data on tcp://127.0.0.1:5555. With the shmem transport and --shm-monitor true, a device starts fairmq-shmmonitor if no monitor runs for the session. The device looks for the monitor in $FAIRMQ_PATH and then in $PATH. The script sets FAIRMQ_PATH to the build directory. The option --shm-throw-bad-alloc false makes the sampler wait for free memory in a full segment.

Run the benchmark with 1 MB messages and the shmem transport, without an iteration limit:

cd build/fairmq
./startMQBenchmark.sh 1000000 0 shmem

The windows stay open after the devices end (xterm -hold).

Related pages

• Run a benchmark: measure throughput with the sampler and the sink.
• Example: built-in devices: all six devices in one topology.
• Command-line options: the options that all devices have.
• Channel properties: sndBufSize, rateLogging and the other properties.
• Monitor and clean up: fairmq-shmmonitor and the option --shm-monitor.
• Design a topology: fan-out and fan-in with the splitter, the merger and the multiplier.