Reference

Helpers in fair::mq::tools

The namespace fair::mq::tools contains small helpers that FairMQ uses in its own code: network lookups, UUIDs, child processes, string functions, a rate limiter and a version type. The installation includes their headers, so your devices and tests can use them. Some helpers are internal in practice: only the tests use them, or only the key input of FairMQ programs.

The installed headers

fairmq/CMakeLists.txt installs twelve headers of fairmq/tools/ and the header fairmq/Tools.h, which includes eleven of them. The table shows where FairMQ uses each header.

Header	Contents	Used by	For your code
`fairmq/Tools.h`	Includes all headers of this table, except `IO.h`.	`Device.h`, `Channel.cxx`, `TransportFactory.cxx`	Use the single headers.
`tools/Strings.h`	`ToString()`, `ToStrVector()`, `StrStartsWith()`, `StrEndsWith()`	Most parts of FairMQ	Use.
`tools/Unique.h`	`Uuid()`, `UuidHash()`	`TransportFactory.cxx`, `fairmq-uuid-gen`, the tests	Use.
`tools/Network.h`	`getHostIPs()`, `getInterfaceIP()`, `getDefaultRouteNetworkInterface()`, `getIpFromHostname()`, `DefaultRouteDetectionError`	`Device.cxx`	Use.
`tools/Process.h`	`execute()`, `execute_result`	The tests of FairMQ	Use in tests.
`tools/RateLimit.h`	`RateLimiter`	`Device.cxx` (`--rate`), `fairmq-bsampler`, the region example	Use.
`tools/Version.h`	`Version`	`Device`, `Plugin`	Use. It is a type of the device and plugin API.
`tools/Exceptions.h`	`CallOnDestruction`	`Device.cxx`, `StateMachine.cxx`	Use.
`tools/Semaphore.h`	`Semaphore`, `SharedSemaphore`	Only `test/tools/`	Internal in practice.
`tools/InstanceLimit.h`	`InstanceLimiter`	Only `test/tools/`	Internal in practice.
`tools/IO.h`	`NonCanonicalInput`	The control plugin, `fairmq-shmmonitor`	Internal in practice.
`tools/Compiler.h`	The macro `__FAIRMQ_ALWAYS_INLINE`	Nothing	Internal.
`tools/CppSTL.h`	No declarations	Nothing	Internal.

Include a header as <fairmq/tools/<Name>.h>, and link your target to FairMQ::FairMQ (see Use FairMQ in a CMake project). The functions of Network.h, Process.h, Semaphore.h and Unique.h are compiled into libfairmq. The other headers need no library.

The test suite Tools in test/tools/ tests most helpers. Run it with ctest -R '^Tools\.' in a build with BUILD_TESTING=ON (see Contributing).

Tools.h

fairmq/Tools.h includes all headers of fairmq/tools/ except IO.h. fairmq/Device.h includes fairmq/Tools.h. Thus a source file of a device has the helpers without an include. Include the header of each helper that you use, so that your code does not depend on the includes of Device.h.

Do not confuse fairmq/tools/Version.h with fairmq/Version.h. The build generates fairmq/Version.h from Version.h.in. It defines macros, for example FAIRMQ_VERSION, FAIRMQ_VERSION_MAJOR and FAIRMQ_GIT_VERSION.

Strings.h

Function	What it does
`ToString(T&&... t) -> std::string`	Writes all arguments with `operator<<` into a `std::stringstream`, without separators, and returns the string.
`ToStrVector(const int argc, charconst argv, const bool dropProgramName = true) -> std::vector<std::string>`	Copies the command-line arguments of `main()` into a vector. With `dropProgramName = true`, the vector does not contain `argv[0]`.
`bool StrStartsWith(std::string const& str, std::string const& start)`	`true` if `str` starts with `start`.
`bool StrEndsWith(std::string const& str, std::string const& end)`	`true` if `str` ends with `end`.

ToString() uses the formatting of the stream. Thus 2.5 gives 2.5, and each type with an operator<< works. FairMQ uses it for the texts of its exceptions and log lines. DeviceRunner uses ToStrVector(). fairmq-shmmonitor --list-all uses StrStartsWith() and StrEndsWith() to find the files of FairMQ.

The tests show the behavior, also for empty strings:

TEST(Strings, ToString)
{
    EXPECT_EQ(ToString(), "");
    EXPECT_EQ(ToString("a", 1, '-', 2.5, string("z")), "a1-2.5z");
}

test/tools/_tools.cxx:118–122

TEST(Strings, StartsAndEndsWith)
{
    EXPECT_TRUE(StrStartsWith("fairmq", "fair"));
    EXPECT_TRUE(StrStartsWith("fairmq", ""));
    EXPECT_FALSE(StrStartsWith("fair", "fairmq"));
    EXPECT_FALSE(StrStartsWith("fairmq", "mq"));
    EXPECT_TRUE(StrEndsWith("fairmq", "mq"));
    EXPECT_TRUE(StrEndsWith("fairmq", ""));
    EXPECT_FALSE(StrEndsWith("mq", "fairmq"));
    EXPECT_FALSE(StrEndsWith("fairmq", "fair"));
}

test/tools/_tools.cxx:124–134

TEST(Strings, ToStrVector)
{
    char prog[] = "prog";
    char a[] = "--a";
    char b[] = "b";
    char* argv[] = {prog, a, b};
    EXPECT_EQ(ToStrVector(3, argv), (vector<string>{"--a", "b"}));
    EXPECT_EQ(ToStrVector(3, argv, false), (vector<string>{"prog", "--a", "b"}));
}

test/tools/_tools.cxx:136–144

Unique.h

Function	What it returns
`std::string Uuid()`	A new random UUID as a string with 36 characters.
`std::size_t UuidHash()`	The hash of a new random UUID.

Each call makes a new UUID with the random generator of Boost.Uuid. UuidHash() returns boost::hash of the UUID, not the hash of the string of an earlier Uuid() call. Unique.cxx makes Boost use the files /dev/*random instead of getrandom() and getentropy().

FairMQ uses these functions here:

TransportFactory::CreateTransportFactory() gives a factory with an empty ID the ID Uuid() (TransportFactory.cxx).
fairmq-uuid-gen prints the result of one of the two functions.
The tests make unique session names and ipc:// addresses with them, for example Configure() and IpcAddress in test/helper/Transport.h.

The test makes 1000 values of each function and checks that they are all different:

TEST(Unique, UuidsAreUnique)
{
    set<string> uuids;
    set<size_t> hashes;
    for (int i = 0; i < 1000; ++i) {
        auto const uuid = Uuid();
        EXPECT_EQ(uuid.size(), 36U) << uuid;
        uuids.insert(uuid);
        hashes.insert(UuidHash());
    }
    EXPECT_EQ(uuids.size(), 1000U);
    EXPECT_EQ(hashes.size(), 1000U);
}

test/tools/_tools.cxx:146–158

Network.h

Function	What it returns	On failure
`getHostIPs()`	A `std::map<std::string, std::string>`: the name of each network interface and its IPv4 address.	Throws `std::runtime_error` if `getifaddrs()` or `getnameinfo()` fails.
`getInterfaceIP(const std::string& interface)`	The IPv4 address of the interface, from `getHostIPs()`.	Returns `""`. For an unknown interface, it logs `Could not find provided network interface: "<name>"!, exiting.` at the severity `error`. The program continues.
`getDefaultRouteNetworkInterface()`	The name of the interface of the default route.	Throws `DefaultRouteDetectionError` if it finds no interface. Returns `""` if `popen()` fails.
`getIpFromHostname(const std::string& hostname)`	The first IPv4 address of the host name.	Returns `""` and logs a warning or an error.

getHostIPs() lists only IPv4 addresses (AF_INET). If an interface has several IPv4 addresses, the map keeps the first one.
On Linux, getDefaultRouteNetworkInterface() reads /proc/net/route and selects the line with the destination 00000000. If it finds none, it runs ip route | grep default | cut -d " " -f 5 | head -n 1. On macOS, it runs route -n get default | grep interface | cut -d ":" -f 2 (Network.cxx).
getIpFromHostname() resolves the name with the TCP resolver of Boost.Asio, for IPv4 only.

The device uses these functions for its channel addresses (Device.cxx):

A binding sub-channel without an address gets the IPv4 address of --network-interface. With the value default, the device asks getDefaultRouteNetworkInterface(). If it throws, the device binds to tcp://*:1. See Configure channels.
The device replaces the host name of a tcp:// endpoint with the result of getIpFromHostname(). See Channel properties.

This program prints the interfaces, the interface of the default route and an address of a host name:

#include <fairmq/tools/Network.h>

#include <iostream>

int main()
{
    for (auto const& [name, ip] : fair::mq::tools::getHostIPs()) {
        std::cout << name << ": " << ip << '\n';
    }

    try {
        auto const interface = fair::mq::tools::getDefaultRouteNetworkInterface();
        std::cout << "default route: " << interface << ", "
                  << fair::mq::tools::getInterfaceIP(interface) << '\n';
    } catch (fair::mq::tools::DefaultRouteDetectionError const& e) {
        std::cout << e.what() << '\n';
    }

    std::cout << "localhost: " << fair::mq::tools::getIpFromHostname("localhost") << '\n';
    return 0;
}

The tests in test/tools/_network.cxx check that localhost gives 127.0.0.1 and that an unknown host name gives an empty string:

TEST(Tools, NetworkIPv4Localhost)
{
    auto const ip = fair::mq::tools::getIpFromHostname("localhost");
    EXPECT_FALSE(ip.empty());
    EXPECT_EQ(ip, "127.0.0.1");
}

test/tools/_network.cxx:26–31

Process.h

execute() starts a command in a child process, waits until it stops, and returns an execute_result. The result has the captured output console_out and the exit code exit_code.

execute_result execute(const std::string& cmd,
                       const std::string& prefix = "",
                       const std::string& input = "",
                       int sig = -1);

execute_result execute(const std::string& cmd,
                       const std::string& prefix,
                       const std::string& input,
                       int sig,
                       std::chrono::milliseconds timeout);

Parameter	Effect
`cmd`	The command line. For shell syntax, for example a redirection, run it with `sh -c`, as the test below does.
`prefix`	The text before each line of the output, for example `"[sampler] "`.
`input`	Text for the standard input of the child. `execute()` writes it 1000 ms after the start. It does not close the standard input after that.
`sig`	A signal, for example `SIGINT`. If it is not `-1`, `execute()` sends it to the child 2000 ms after the start.
`timeout`	Second overload only. After this time, `execute()` stops the child with `SIGKILL`. `0` means no limit. The first overload has no limit.

The overload with timeout is on dev and is not in a release yet (see Release notes).

execute() writes each line to std::cout and to console_out, with the prefix (Process.cxx):

first the command, then fair::mq::tools::execute: pid: <pid>,
each line of the standard output of the child,
each line of the standard error, after the text error: ,
fair::mq::tools::execute: timeout after <n> ms, killing pid <pid> if the timeout ends,
at the end, fair::mq::tools::execute: exit code: <n>.

Since v1.11.0, calls from several threads write whole lines to std::cout. The test helper RunDevices() in test/helper/Devices.h starts each device of a test with execute() in its own thread. See Test your devices.

The tests show the output, the exit code, the signal and the timeout. When a signal stops the child, the tests expect the number of the signal as exit_code, for example SIGKILL after the timeout.

TEST(Execute, OutputAndExitCode)
{
    auto const result = execute(R"(sh -c "echo out; echo err 1>&2; exit 3")", "[p] ");
    EXPECT_EQ(result.exit_code, 3);
    EXPECT_NE(result.console_out.find("[p] out\n"), string::npos) << result.console_out;
    EXPECT_NE(result.console_out.find("[p] error: err\n"), string::npos) << result.console_out;
}

test/tools/_tools.cxx:160–166

TEST(Execute, Timeout)
{
    auto const start = chrono::steady_clock::now();
    auto const result = execute("sleep 30", "", "", -1, chrono::milliseconds(200));
    EXPECT_LT(chrono::steady_clock::now() - start, chrono::seconds(10));
    EXPECT_EQ(result.exit_code, SIGKILL);
    EXPECT_NE(result.console_out.find("timeout after 200 ms"), string::npos) << result.console_out;
}

test/tools/_tools.cxx:182–189

RateLimit.h

RateLimiter limits a loop to a number of iterations per second.

Member	What it does
`explicit RateLimiter(float rate)`	Sets the rate in Hz. A rate of 0 or less sets 1 GHz, which means no limit in practice.
`void maybe_sleep()`	Call it at the end of each iteration. It sleeps if the iterations come too fast.

maybe_sleep() reads the clock only every n-th call. It measures the mean time of an iteration, and it changes n and the sleep time until the time agrees with the rate (RateLimit.h). The device uses a RateLimiter for --rate in its ConditionalRun() loop, and only for rates above 0.001:

tools::RateLimiter rateLimiter(fRate);

while (!NewStatePending() && ConditionalRun()) {
    if (fRate > 0.001) {
        rateLimiter.maybe_sleep();
    }
}

fairmq/Device.cxx:413–419

fairmq-bsampler uses it for --msg-rate (see Built-in devices). The test checks a low lower limit of the time (350 ms for 0.5 s), because a slow machine makes the loop slower, never faster. The upper limit is 5 s:

TEST(RateLimiter, LimitsTheRate)
{
    // 100 calls at 200 Hz take at least approx. 0.5 s. Check only a loose
    // lower bound: a slow machine can make the loop slower, never faster.
    RateLimiter limiter(200);
    auto const start = chrono::steady_clock::now();
    for (int i = 0; i < 100; ++i) {
        limiter.maybe_sleep();
    }
    auto const elapsed = chrono::steady_clock::now() - start;
    EXPECT_GE(elapsed, chrono::milliseconds(350));
    EXPECT_LT(elapsed, chrono::seconds(5));
}

test/tools/_tools.cxx:34–46

The test RateLimiter.ZeroRateDoesNotSleep checks that 100000 calls with the rate 0 take less than 1 s.

Version.h

Version holds three numbers: fkMajor, fkMinor and fkPatch, all const int. Make it with braces, for example fair::mq::tools::Version{1, 2, 0}.

The operators <, >, <=, >=, == and != compare the major, then the minor, then the patch number.
operator<< writes 1.2.0.

The type is part of the device and plugin API:

The constructors Device(tools::Version) and Device(ProgOptions&, tools::Version) set the version of your device. The default is {0, 0, 0}. GetVersion() returns it. DeviceRunner logs it for --version. Today --version prints nothing: this is a known defect, see the disabled tests in test/device/_runner.cxx.
Plugin::Version is an alias of tools::Version. See Write a plugin.

TEST(PluginVersion, Operators)
{
    struct tools::Version v1{1, 0, 0};
    struct tools::Version v2{1, 0, 0};
    struct tools::Version v3{1, 2, 0};
    EXPECT_EQ(v1, v2);
    EXPECT_NE(v1, v3);
    EXPECT_GT(v3, v2);
    EXPECT_LT(v1, v3);
    EXPECT_GE(v3, v2);
    EXPECT_GE(v2, v1);
    EXPECT_LE(v1, v2);
    EXPECT_LE(v2, v3);
}

test/plugins/_plugin.cxx:70–83

Exceptions.h

CallOnDestruction calls a function in its destructor. Use it to run code when a scope ends with an exception, without a catch block.

Member	What it does
`CallOnDestruction(std::function<void()> c, bool enable = true)`	Stores the function.
`enable()`, `disable()`	Switch the call on or off.
`~CallOnDestruction()`	Calls the function if the object is enabled.

The device uses it in the state RUNNING. If PreRun(), ConditionalRun(), Run(), PostRun() or a data callback throws, the destructor requests the transition ERROR FOUND. At the normal end of the function, cod.disable() stops the call (Device.cxx):

tools::CallOnDestruction cod([&](){
    ChangeStateOrThrow(Transition::ErrorFound);
});

fairmq/Device.cxx:398–400

StateMachine::ProcessWork() uses the same pattern to go to ERROR (StateMachine.cxx). A destructor is noexcept by default in C++. Thus a callback that throws ends the program with std::terminate(). No test of test/tools/ covers this class. The tests of the ERROR state in test/device/ show its effect.

Semaphore.h

Internal in practice: the library does not use these classes.

Class	Description
`Semaphore`	A counting semaphore with a mutex and a condition variable. Not copyable.
`SharedSemaphore`	Copyable. All copies share one `Semaphore` through a `std::shared_ptr`.

Both classes have the same members: Semaphore() starts with the count 0, and explicit Semaphore(std::size_t initial_count) with the given count. Wait() blocks until the count is above 0, then decrements it. Signal() increments the count and wakes one waiting thread. GetCount() returns the count.

Wait() has no timeout. For tests, the test helper test::Semaphore in test/helper/Sync.h has a Wait() with a deadline, so that a missing signal does not block the test. An installed FairMQ does not contain the test helpers. Copy the file into your project if you need it.

TEST(SharedSemaphore, CopiesShareTheCount)
{
    SharedSemaphore sem(1);
    SharedSemaphore copy = sem; // NOLINT(performance-unnecessary-copy-initialization)
    copy.Signal();
    EXPECT_EQ(sem.GetCount(), 2U);
    sem.Wait();
    EXPECT_EQ(copy.GetCount(), 1U);
}

test/tools/_tools.cxx:108–116

InstanceLimit.h

Internal in practice: the library does not use this class.

InstanceLimiter<Tag, Max> counts its objects. The constructor increments a static counter. If Max objects exist already, it throws std::runtime_error with the text More than <Max> instances of <Tag> in parallel not supported. The destructor decrements the counter. GetCount() returns it.

Each pair of Tag and Max has its own counter.
Tag must be default-constructible and must have an operator<<, because the text of the exception contains Tag().
The counter is a plain static int without a lock (InstanceLimit.h).

In the test, TestTag writes TestTag with its operator<<:

TEST(InstanceLimiter, ThrowsAboveTheLimit)
{
    using Limiter = InstanceLimiter<TestTag, 2>;
    {
        Limiter first;
        Limiter second;
        EXPECT_EQ(first.GetCount(), 2);
        try {
            Limiter third;
            ADD_FAILURE() << "no exception for the third instance";
        } catch (runtime_error const& e) {
            EXPECT_EQ(string(e.what()), "More than 2 instances of TestTag in parallel not supported");
        }
        EXPECT_EQ(first.GetCount(), 2);
    }
    Limiter again;
    EXPECT_EQ(again.GetCount(), 1);
}

test/tools/_tools.cxx:63–80

IO.h

Internal in practice: the control plugin uses it for the keys of the interactive mode (Control.cxx), and fairmq-shmmonitor -i for its keys (Monitor.cxx). fairmq/Tools.h does not include this header.

The constructor of NonCanonicalInput switches off the flags ICANON and ECHO of the terminal on the standard input. Then a program gets each key at once, and the terminal does not show it. The destructor switches both flags on again. It does not restore the earlier settings. The class is not copyable and not movable. No test covers it.

Compiler.h and CppSTL.h

Both headers are internal.

Compiler.h defines the macro __FAIRMQ_ALWAYS_INLINE as __attribute__((always_inline)) inline. No file of the repository uses it.
CppSTL.h contains only the empty namespace fair::mq::tools.

fairmq-uuid-gen

fairmq-uuid-gen prints a new random UUID, or the hash of a new UUID, and stops. It calls Uuid() or UuidHash(). The build makes it in the directory fairmq of the build tree. The installation puts it into <prefix>/bin. The imported target is FairMQ::fairmq-uuid-gen (see Use FairMQ in a CMake project).

Option	Default	Description
`-h, --hash`bool	no value: true	Generates UUID and returns its hash.runUuidGenerator.cxx
`--help`flag	—	Print helprunUuidGenerator.cxx
No option matches. Clear the filter or choose All.

Command	Output
`fairmq-uuid-gen`	A UUID with 36 characters.
`fairmq-uuid-gen -h`	The hash as a decimal number.
`fairmq-uuid-gen --help`	`UUID generator` and the options.

The output is one line on the standard output:

if (hash)
{
    std::cout << fair::mq::tools::UuidHash() << std::endl;
}
else
{
    std::cout << fair::mq::tools::Uuid() << std::endl;
}

return 0;

fairmq/tools/runUuidGenerator.cxx:41–50

The exit code is 0. For a wrong option, the tool writes Unhandled Exception reached the top of main: <text>, application will now exit to the standard error and returns 2.

The test scripts of the examples make a new session name for each run with -h. They also use the session name in the ipc:// addresses, so that two tests at the same time do not use the same shared memory or the same socket files:

session="$(@CMAKE_BINARY_DIR@/fairmq/fairmq-uuid-gen -h)"
chan="data"
chanAddr="/tmp/fmq_$session""_""$chan""_""$transport"

examples/1-1/test-ex-1-1.sh.in:13–15

The hash has only digits, so it can be part of a file name. Test your devices shows a complete test script.

Test your devices: fairmq-uuid-gen and execute() in tests.
Receive data with callbacks or loops: --rate and the RateLimiter.
Configure channels: addresses, network interfaces and host names.
Sessions and segments: why each test needs its own session.
API reference of fair::mq::tools