Guides

Control a device from code

A controller requests the transitions of a device. This guide shows the ways to write your own controller, also without DeviceRunner. It also shows how to read and change properties while the device runs.

Choose a way

Way	Use it when
A controller plugin	The device starts with `runDevice.h` or `DeviceRunner`, and no mode of `--control` fits.
A plugin in your program	You want one executable that contains the device and its controller.
A device in a thread	Your program creates the device without `DeviceRunner`, for example in a test.
Your own `main()` with plugins	You want the plugins and the command line of a device, but not `DeviceRunner`.

All ways use the same states and transitions. See Device states and States and transitions.

Control the device from a plugin

A plugin requests a transition with ChangeDeviceState(). Only the plugin with the controller role can do this. See The controller role.

Write a plugin. See Write a plugin.
In the constructor, call TakeDeviceControl(). The plugins of -P are created before the control plugin. Thus your plugin gets the role first.
In the constructor, subscribe to the state changes with SubscribeToDeviceStateChange().
In the callback, request the next transition in each idling state. In INITIALIZING DEVICE, request CompleteInit.
In EXITING, call ReleaseDeviceControl(). DeviceRunner::Run() returns only after the controller releases the role.
Call ReleaseDeviceControl() in the destructor too. Without the role, the call does nothing.

The plugin of the custom-controller example does these steps:

struct MyController : fair::mq::Plugin   // NOLINT
{
    template<typename... Args>
    MyController(Args&&... args)
        : Plugin(std::forward<Args>(args)...)
    {
        TakeDeviceControl();

        SubscribeToDeviceStateChange([this](auto state) {
            auto shutdown = GetProperty<bool>("please-shut-me-down", false);
            try {
                switch (state) {
                    case DeviceState::Idle: {
                        ChangeDeviceState(shutdown ? DeviceStateTransition::End
                                                    : DeviceStateTransition::InitDevice);
                        break;
                    }
                    case DeviceState::InitializingDevice: {
                        ChangeDeviceState(DeviceStateTransition::CompleteInit);
                        break;
                    }
                    case DeviceState::Initialized: {
                        ChangeDeviceState(DeviceStateTransition::Bind);
                        break;
                    }
                    case DeviceState::Bound: {
                        ChangeDeviceState(DeviceStateTransition::Connect);
                        break;
                    }
                    case DeviceState::DeviceReady: {
                        ChangeDeviceState(shutdown ? DeviceStateTransition::ResetDevice
                                                    : DeviceStateTransition::InitTask);
                        break;
                    }
                    case DeviceState::Ready: {
                        ChangeDeviceState(shutdown ? DeviceStateTransition::ResetTask
                                                    : DeviceStateTransition::Run);
                        break;
                    }
                    case DeviceState::Running: {
                        ChangeDeviceState(DeviceStateTransition::Stop);
                        break;
                    }
                    case DeviceState::Exiting: {
                        ReleaseDeviceControl();
                        break;
                    }
                    default:
                        break;
                }
            } catch (fair::mq::PluginServices::DeviceControlError const&) {
                // this means we do not have device control
            }
        });
    }

    ~MyController() override { ReleaseDeviceControl(); }
};

examples/custom-controller/MyController.h:17–74

The control plugin then cannot take the role. It logs the reason with the severity debug and starts no controller. Its signal handling still works.

Rules for the state callback

The callback runs in the thread of the state machine, at the start of each state. The device calls the hooks of the state after all state callbacks return. See Threads in a device.
Request no transition in BINDING, CONNECTING, INITIALIZING TASK, RESETTING TASK and RESETTING DEVICE. The device ends these states by itself with the transition Auto.
INITIALIZING DEVICE waits until the controller requests CompleteInit. Set the configuration of the device before this request.
The device ends RUNNING by itself when its work ends: it requests Stop. A Stop from the callback of RUNNING is pending before the work starts. Then ConditionalRun() and the data callbacks do not run. PreRun() and PostRun() still run. Run() also runs, but only in a device without data callbacks.
ChangeDeviceState() returns false when the state machine refuses the transition.
ChangeDeviceState() throws fair::mq::PluginServices::DeviceControlError when another plugin has the role. If no plugin has the role, the call gives the role to the calling plugin.
An exception that leaves the callback moves the device to ERROR. Catch the exceptions in the callback.

A controller with its own thread

The device waits until the state callback returns. For a controller that waits, for example for a timer or for input, use a thread. The built-in control plugin works so:

Its state callback puts each new state into a fair::mq::StateQueue.
Its controller thread requests a transition. Then it waits in the queue for the next state.

SubscribeToDeviceStateChange([&](DeviceState newState) {
    LOG(trace) << "control plugin notified on new state: " << newState;

    if (newState == DeviceState::Error) {
        fPluginShutdownRequested = true;
        fStateQueue.Push(newState, [this]{ fDeviceShutdownRequested = true; });
    } else {
        fStateQueue.Push(newState);
    }
});

fairmq/plugins/control/Control.cxx:56–65

auto Control::RunStartupSequence() -> void
{
    using Transition = DeviceStateTransition;
    using State = DeviceState;
    auto shutdownRequested = [this]{ return fDeviceShutdownRequested.load(); };

    ChangeDeviceState(Transition::InitDevice);
    fStateQueue.WaitForStateOrCustom(State::InitializingDevice, shutdownRequested);
    if (fDeviceShutdownRequested) { return; /* --> shutdown sequence */ }

    ChangeDeviceState(Transition::CompleteInit);
    fStateQueue.WaitForStateOrCustom(State::Initialized,        shutdownRequested);
    if (fDeviceShutdownRequested) { return; /* --> shutdown sequence */ }

    ChangeDeviceState(Transition::Bind);
    fStateQueue.WaitForStateOrCustom(State::Binding,            shutdownRequested);
    if (fDeviceShutdownRequested) { return; /* --> shutdown sequence */ }

    fStateQueue.WaitForStateOrCustom(State::Bound,              shutdownRequested);
    if (fDeviceShutdownRequested) { return; /* --> shutdown sequence */ }

    ChangeDeviceState(Transition::Connect);
    fStateQueue.WaitForStateOrCustom(State::Connecting,         shutdownRequested);
    if (fDeviceShutdownRequested) { return; /* --> shutdown sequence */ }

    fStateQueue.WaitForStateOrCustom(State::DeviceReady,        shutdownRequested);
    if (fDeviceShutdownRequested) { return; /* --> shutdown sequence */ }

    ChangeDeviceState(Transition::InitTask);
    fStateQueue.WaitForStateOrCustom(State::InitializingTask,   shutdownRequested);
    if (fDeviceShutdownRequested) { return; /* --> shutdown sequence */ }

    fStateQueue.WaitForStateOrCustom(State::Ready,              shutdownRequested);
    if (fDeviceShutdownRequested) { return; /* --> shutdown sequence */ }

    ChangeDeviceState(Transition::Run);
    fStateQueue.WaitForStateOrCustom(State::Running,            shutdownRequested);
    if (fDeviceShutdownRequested) { return; /* --> shutdown sequence */ }
}

fairmq/plugins/control/Control.cxx:104–142

WaitForState(state) takes states from the queue until it gets state.
WaitForStateOrCustom(state, predicate) also returns when the predicate is true. The queue checks the predicate after each Push() and Notify().
WaitForNext() returns the next state of the queue.
When the next state of the queue is ERROR, these functions throw fair::mq::DeviceErrorState.

Put the controller in your program

A plugin can be part of the executable. The custom-controller example works so.

Put the plugin class and the macro REGISTER_FAIRMQ_PLUGIN into the code of your program.
In main(), create a fair::mq::DeviceRunner with argc and argv.
In the hook hooks::LoadPlugins, load the plugin with the prefix s: and its name, for example s:mycontroller.
In the hook hooks::InstantiateDevice, create the device with r.fConfig and put it into r.fDevice.
Return the result of RunWithExceptionHandlers().
In CMake, set the target property ENABLE_EXPORTS to ON. The plugin manager finds the plugin in the symbols of the executable.

#include "MyController.h"
#include "MyDevice.h"

#include <chrono>   // for std::chrono_literals
#include <fairlogger/Logger.h>
#include <fairmq/DeviceRunner.h>
#include <memory>   // for std::make_unique

int main(int argc, char* argv[])
{
    using namespace fair::mq;
    using namespace std::chrono_literals;

    DeviceRunner runner(argc, argv);

    runner.AddHook<hooks::LoadPlugins>([](DeviceRunner& r) {
        r.fPluginManager.LoadPlugin("s:mycontroller");
        // 's:' stands for static because the plugin is compiled into the executable
        // 'mycontroller' is the plugin name passed as second arg to REGISTER_FAIRMQ_PLUGIN
    });

    fair::Logger::SetConsoleSeverity(fair::Severity::debug);

    runner.AddHook<hooks::InstantiateDevice>([](DeviceRunner& r) {
        r.fConfig.SetProperty<int>("catch-signals", 0);
        r.fConfig.SetProperty<bool>("please-shut-me-down", false);
        r.fDevice = std::make_unique<example::MyDevice>(r.fConfig);

        r.fDevice->SubscribeToStateChange("example", [&r](auto state) {
            if (state == State::Running) {
                r.fDevice->WaitFor(3s);
                r.fConfig.SetProperty<bool>("please-shut-me-down", true);
            }
        });
    });

    return runner.RunWithExceptionHandlers();
}

examples/custom-controller/main.cxx:9–46

add_executable(${exe} main.cxx)
target_link_libraries(${exe} PRIVATE FairMQ)
set_target_properties(${exe} PROPERTIES ENABLE_EXPORTS ON)

examples/custom-controller/CMakeLists.txt:11–13

DeviceRunner keeps one function for each type of hook. A second AddHook() of the same type replaces the first. The order of the hooks is in How a device starts.

Run the device in a thread

Without DeviceRunner, the device has no config plugin and no control plugin. Your program sets the configuration and requests all transitions.

Create the device: a fair::mq::Device or an object of your device class.
Configure the device, for example with SetTransport() and AddChannel().
Start RunStateMachine() in a thread. It returns when the device reaches EXITING.
Request each transition with ChangeStateOrThrow().
Wait for the next idling state with WaitForState().
In IDLE, request End. Then join the thread.

The test Config.SetTransport does these steps:

string TestDeviceSetTransport(const string& transport)
{
    Device device;
    device.SetTransport(transport);

    Channel channel;
    channel.UpdateType("pub");
    channel.UpdateMethod("connect");
    channel.UpdateAddress("tcp://localhost:5558");
    device.AddChannel("data", std::move(channel));

    thread t(&Device::RunStateMachine, &device);

    test::Control(device);

    if (t.joinable()) {
        t.join();
    }

    return device.GetTransportName();
}

test/device/_config.cxx:135–155

The test helper Control() requests the transitions of a full cycle. The device of the test has no own Run() code. It goes back to READY by itself, so Control() waits for READY after RUNNING.

inline auto Control(Device& device, Cycle cycle = Cycle::FullNoStop)
{
    if(   cycle == Cycle::FullNoStop
       || cycle == Cycle::ToDeviceReadyAndBack
       || cycle == Cycle::ToRun)
    {
        device.ChangeStateOrThrow(Transition::InitDevice);
        device.WaitForState(State::InitializingDevice);
        device.ChangeStateOrThrow(Transition::CompleteInit);
        device.WaitForState(State::Initialized);
        device.ChangeStateOrThrow(Transition::Bind);
        device.WaitForState(State::Bound);
        device.ChangeStateOrThrow(Transition::Connect);
        device.WaitForState(State::DeviceReady);
    }
    if(   cycle == Cycle::FullNoStop
       || cycle == Cycle::ToRun)
    {
        device.ChangeStateOrThrow(Transition::InitTask);
        device.WaitForState(State::Ready);
        device.ChangeStateOrThrow(Transition::Run);
        device.WaitForState(State::Running);
    }
    if(   cycle == Cycle::FullNoStop
       || cycle == Cycle::ReadyToEnd)
    {
        device.WaitForState(State::Ready);
        device.ChangeStateOrThrow(Transition::ResetTask);
        device.WaitForState(State::DeviceReady);
    }
    if(   cycle == Cycle::FullNoStop
       || cycle == Cycle::ToDeviceReadyAndBack
       || cycle == Cycle::ReadyToEnd)
    {
        device.ChangeStateOrThrow(Transition::ResetDevice);
        device.WaitForState(State::Idle);
        device.ChangeStateOrThrow(Transition::End);
    }
}

test/helper/ControlDevice.h:28–66

The configuration of the device

Without a ProgOptions object, the device uses its own internal object.
To use your object, give it to the constructor Device(ProgOptions&) or to SetConfig(). SetConfig() deletes the internal object. Thus call it before the other setters.
Set the device ID with the property id. SetId() has no effect on a device that you initialize, because INITIALIZING DEVICE reads the property id.

The functions of the device

Function	What it does
`ChangeState(transition)`	Requests a transition. Returns `false` when the current state does not accept it, or when another transition is pending. `ChangeState(name)` takes the name, for example `"INIT DEVICE"`.
`ChangeStateOrThrow()`	Requests a transition. Throws `std::system_error` when `ChangeState()` returns `false`. The text starts with the name of the transition, for example `Invalid transition: CONNECT`. The error code is `ErrorCode::DeviceChangeStateFailed`.
`WaitForState(state)`	Waits until the device enters `state`. `WaitForState(name)` takes the name, for example `"READY"`.
`WaitForNextState()`	Waits for the next state and returns it.
`GetCurrentState()`, `GetCurrentStateName()`	Return the current state.
`SubscribeToStateChange(key, callback)`	Calls `callback` at the start of each state, in the thread of the state machine. `UnsubscribeFromStateChange(key)` removes it.
`SubscribeToNewTransition(key, callback)`	Calls `callback` when the state machine accepts a transition, in the thread that requests the transition.
`NewStatePending()`	Returns `true` when a transition is requested and the next state has not started yet.

ChangeState() is [[nodiscard]]. A name that does not exist makes ChangeState(name) and WaitForState(name) throw std::out_of_range. The names are in States and transitions.

How WaitForState() works

The device puts each new state into a queue. WaitForState() takes states from the queue until it finds the given state. Thus it returns at once for a state that is in the queue already.

The first IDLE is not in the queue. Do not wait for it.
After the transition ErrorFound, the queue gets ERROR. Then WaitForState() and WaitForNextState() throw fair::mq::DeviceErrorState with the text Controlled device transitioned to error state.
After an exception in a hook, the queue gets no ERROR. Then WaitForState() does not return.

RunStateMachine() lets the exception of a hook pass to its caller. Catch it in the thread: an exception that leaves the function of a std::thread calls std::terminate(). To see ERROR in all cases, use SubscribeToStateChange(). The state machine calls the subscribers with ERROR also after an exception.

The class StateMachine

Each device has a private fair::mq::StateMachine. Use the functions of the device for it. The class also works without a device. The tests in test/state_machine/_state_machine.cxx use it so.

Function	What it does
`Start()`	Starts the state machine in `IDLE`. Call it before the first transition.
`ProcessWork()`	Runs the loop. It first calls the subscribers with `IDLE`. For each new state, it calls the `PrepareState()` function, the subscribers and the `HandleStates()` function. It returns in `EXITING`.
`PrepareState(f)`, `HandleStates(f)`	Set the functions for each new state. Each takes one function. A second call logs an error. `StopHandlingStates()` removes both.
`WaitForPendingState()`, `WaitForPendingStateFor(ms)`	Wait until a transition is requested. `WaitForPendingStateFor()` returns `false` after the timeout.
`ChangeState()`, `SubscribeToStateChange()`, `SubscribeToNewTransition()`, `NewStatePending()`, `GetCurrentState()`	Work as the functions of the device.

In ERROR, ProcessWork() throws StateMachine::ErrorStateException. When a function of the loop throws, ProcessWork() sets ERROR, calls the subscribers with ERROR, and passes the exception on.

Parse options without DeviceRunner

Without DeviceRunner, no code reads the command line. Your program can parse it with a fair::mq::ProgOptions object and give this object to the device.

Create a ProgOptions object. It knows only the general options: --help, --version, --severity, --file-severity, --verbosity, --color, --log-to-file and --print-options (ProgOptions.cxx).
Add your options with AddToCmdLineOptions(). It takes a boost::program_options::options_description. Add all options before the parse.
Call ParseAll(argc, argv, true).
To apply the log options, call DeviceRunner::HandleGeneralOptions(config, false). It is static. For --help and --print-options, it prints the options and returns false. The second argument true prints the FairMQ logo.
Call Notify(). It calls boost::program_options::notify(). DeviceRunner calls it after HandleGeneralOptions() too.
Give the object to the device: use the constructor Device(ProgOptions&) or SetConfig(). See The configuration of the device.
Request the transitions as in Run the device in a thread.

The test Config.SetConfig parses two options and gives the object to the device:

string TestDeviceSetConfig(const string& transport)
{
    ProgOptions config;

    vector<string> emptyArgs = {"dummy", "--id", "test", "--color", "false"};

    config.ParseAll(emptyArgs, true);

    config.SetProperty("transport", transport);

    Device device;
    device.SetConfig(config);

    Channel channel;
    channel.UpdateType("pub");
    channel.UpdateMethod("connect");
    channel.UpdateAddress("tcp://localhost:5558");
    device.AddChannel("data", std::move(channel));

    thread t([&]() { test::Control(device); });

    device.RunStateMachine();

    if (t.joinable()) {
        t.join();
    }

    return device.GetTransportName();
}

test/device/_config.cxx:58–86

ParseAll() first removes all properties, also the values of SetProperty() (ProgOptions.cxx). Then it stores the values of the command line and the defaults of the registered options. Call SetProperty() after ParseAll(), as the test does.
The last argument true accepts options that are not registered. ParseAll() does not store them as properties. PrintOptions() logs each of them at the severity debug: detected unregistered option: …. In the test above, --id is not registered, so the device ID stays empty.
With false, an option that is not registered throws boost::program_options::unknown_option. The test in test/properties/_properties.cxx checks this.
The first element of argv, or of the vector of the other overload, is the name of the program. The parser skips it (Boost behavior).

The device options and the channels

The device options, for example --id, --transport, --session, --mq-config and --channel-config, are the options of the config plugin (Config.cxx). In INITIALIZING DEVICE, the config plugin also turns --mq-config or --channel-config into channel properties (Config.cxx). A program without DeviceRunner has two ways to get them:

Load the config plugin with a plugin manager, and add its options before the parse. See Load plugins without DeviceRunner. The test Config.SetConfigWithPlugins does this:

string TestDeviceSetConfigWithPlugins(const string& transport)
{
    ProgOptions config;

    vector<string> emptyArgs = {"dummy", "--id", "test", "--color", "false"};

    config.SetProperty("transport", transport);

    Device device;
    PluginManager mgr;
    mgr.LoadPlugin("s:config");
    mgr.ForEachPluginProgOptions([&](boost::program_options::options_description options) {
        config.AddToCmdLineOptions(options);
    });
    mgr.EmplacePluginServices(config, device);
    mgr.InstantiatePlugins();

    config.ParseAll(emptyArgs, true);
    DeviceRunner::HandleGeneralOptions(config);
    device.SetConfig(config);

    Channel channel;
    channel.UpdateType("pub");
    channel.UpdateMethod("connect");
    channel.UpdateAddress("tcp://localhost:5558");
    device.AddChannel("data", std::move(channel));

    thread t([&]() { test::Control(device); });

    device.RunStateMachine();

    config.PrintOptions();

    if (t.joinable()) {
        t.join();
    }

    return device.GetTransportName();
}

test/device/_config.cxx:88–126

Register only the options that you need, and make the channel properties yourself. SuboptParser() turns the values of --channel-config into channel properties. JSONParser() reads them from a JSON file. SetProperties() stores them. Store them before you request CompleteInit, because the device reads the channel properties after this request.

This function does the second way for the options --id and --channel-config:

#include <fairmq/ProgOptions.h>
#include <fairmq/SuboptParser.h>

#include <boost/program_options.hpp>

#include <string>
#include <vector>

void Configure(fair::mq::ProgOptions& config, int argc, char* argv[])
{
    namespace bpo = boost::program_options;
    bpo::options_description options("My device");
    options.add_options()
        ("id", bpo::value<std::string>()->default_value("sampler1"), "Device ID")
        ("channel-config", bpo::value<std::vector<std::string>>()->multitoken()->composing(), "Channels");
    config.AddToCmdLineOptions(options);

    config.ParseAll(argc, argv, true);
    config.Notify();

    if (config.Count("channel-config")) {
        auto const id = config.GetProperty<std::string>("id");
        auto const channels = config.GetProperty<std::vector<std::string>>("channel-config");
        config.SetProperties(fair::mq::SuboptParser(channels, id));
    }
}

For a value without name, SuboptParser() throws boost::property_tree::ptree_bad_path. For an empty device ID, it throws fair::mq::ParserError. The test SuboptParser.Errors checks both. The device options that you do not register, for example transport and rate, have the defaults of the device.

Load plugins without DeviceRunner

A fair::mq::PluginManager loads, creates and owns the plugins of a device. Without DeviceRunner, your program must call its functions in the sequence of DeviceRunner::Run():

Step	Call	Notes
1	`LoadPlugin(name)` or `LoadPlugins(names)`	Finds the plugin and its options. It does not create the plugin. `DeviceRunner` loads the plugins of `-P` (in the constructor of the plugin manager), then `s:config`, then the plugins of the hook `LoadPlugins`, then `s:control`. For the prefixes, see Load a plugin.
2	`ForEachPluginProgOptions()` with `AddToCmdLineOptions()`	Adds the options of the loaded plugins. `DeviceRunner` also adds `PluginManager::ProgramOptions()` (`-P` and `-S`).
3	`ParseAll()`, `HandleGeneralOptions()`, `Notify()`	See Parse options without DeviceRunner.
4	`SetConfig()` of the device	Give the device the `ProgOptions` object of the plugins. Otherwise the device does not see the properties that the plugins set, for example the channels of the config plugin.
5	`EmplacePluginServices(config, device)`	Makes the `PluginServices` object, which connects the plugins to the device and to the properties.
6	`InstantiatePlugins()`	Creates the plugins in the load sequence. The constructors run now, and a controller takes its role.
7	`RunStateMachine()` of the device	Returns in `EXITING`.
8	`WaitForPluginsToReleaseDeviceControl()`	Waits until no plugin has the controller role.

This main() has the minimum sequence for the built-in plugins. MyDevice is your device class:

#include <fairmq/Device.h>
#include <fairmq/DeviceRunner.h>   // DeviceRunner::HandleGeneralOptions()
#include <fairmq/PluginManager.h>
#include <fairmq/ProgOptions.h>
#include <fairmq/tools/Strings.h>  // tools::ToStrVector()

#include <boost/program_options.hpp>

int main(int argc, char* argv[])
{
    // Declare the configuration and the device before the plugin manager.
    // Then the plugin manager and its plugins are destroyed first.
    fair::mq::ProgOptions config;
    MyDevice device;
    fair::mq::PluginManager mgr(fair::mq::tools::ToStrVector(argc, argv, false)); // reads -S and -P

    mgr.LoadPlugin("s:config");
    mgr.LoadPlugin("s:control");
    mgr.ForEachPluginProgOptions([&](boost::program_options::options_description options) {
        config.AddToCmdLineOptions(options);
    });
    config.AddToCmdLineOptions(fair::mq::PluginManager::ProgramOptions());

    config.ParseAll(argc, argv, true);
    if (!fair::mq::DeviceRunner::HandleGeneralOptions(config, false)) {
        return 0; // --help or --print-options
    }
    config.Notify();

    device.SetConfig(config);
    mgr.EmplacePluginServices(config, device);
    mgr.InstantiatePlugins();

    device.RunStateMachine();
    mgr.WaitForPluginsToReleaseDeviceControl();
    return 0;
}

The test PluginManager.LoadPluginStatic uses the control plugin without a parse. It sets the two options of the control plugin with SetProperty() before InstantiatePlugins():

TEST(PluginManager, LoadPluginStatic)
{
    Device device;
    PluginManager mgr;
    device.SetTransport("zeromq");

    ASSERT_NO_THROW(mgr.LoadPlugin("s:control"));

    ProgOptions config;
    config.SetProperty<string>("control", "static");
    config.SetProperty("catch-signals", 0);
    mgr.EmplacePluginServices(config, device);

    ASSERT_NO_THROW(mgr.InstantiatePlugins());

    // check order
    const auto expected = vector<string>{"control"};
    auto actual = vector<string>();
    mgr.ForEachPlugin([&](Plugin& plugin){ actual.push_back(plugin.GetName()); });
    ASSERT_EQ(actual, expected);

    // program options
    auto count = 0;
    mgr.ForEachPluginProgOptions([&count](const options_description&){ ++count; });
    ASSERT_EQ(count, 1);

    device.RunStateMachine();

    mgr.WaitForPluginsToReleaseDeviceControl();
}

test/plugins/_plugin_manager.cxx:72–101

Rules for the sequence:

Call EmplacePluginServices() before InstantiatePlugins() and before WaitForPluginsToReleaseDeviceControl(). Both use the PluginServices object, and the plugin manager does not check that it exists.
The plugins read their options in their constructors. The control plugin reads control and catch-signals. If a property is missing, InstantiatePlugins() throws PluginManager::PluginInstantiationError, for example An error occurred while instantiating plugin control: Config has no key: control.
The plugins unsubscribe from the device in their destructors. Thus the device must exist until the plugin manager is destroyed.
The constructor PluginManager(args) reads -S and -P from args and loads the plugins of -P. It throws PluginManager::ProgramOptionsParseError for an option value that it cannot parse. The default constructor reads nothing.
With the default --control dynamic, the control plugin reads keys from the terminal. For a program without a terminal, use --control static. See Control modes.

Several devices in one process

One process can run more than one device. The test MultipleDevices.TwoInSameProcess runs a sender and a receiver in one process. Each device runs RunStateMachine() in its own thread, and its controller requests the transitions from another thread:

class MultipleDevices : public ::testing::Test {
  public:
    test::IpcAddress const fAddress{"multiple_devices"};

    bool TestFirst()
    {
        test::Sender sender("data");
        sender.SetTransport("zeromq");

        Channel channel("push", "connect", fAddress);
        channel.UpdateRateLogging(0);
        sender.AddChannel("data", std::move(channel));

        control(sender);
        return true;
    }

    bool TestSecond()
    {
        test::Receiver receiver("data");
        receiver.SetTransport("zeromq");

        Channel channel("pull", "bind", fAddress);
        channel.UpdateRateLogging(0);
        receiver.AddChannel("data", std::move(channel));

        control(receiver);
        return true;
    }
};

test/device/_multiple_devices.cxx:38–67

Each device has its own objects:

A state machine. Each Device object has a private StateMachine. The transitions of one device do not change the other devices.
A property store. Without SetConfig(), each device uses its own internal ProgOptions object. If you give your own object, give each device its own. The device reads id and the chans. properties from it, and it writes the bound addresses back.
Transports. In INITIALIZING DEVICE, AddTransport() makes one transport factory for each transport type of the device, from the property store of the device (Device.cxx). All channels of the device with this type use it.

Two devices do not share a transport factory. No constructor of Device takes a factory, and SetTransport() only sets the property transport (the name). The public API has no function that gives a device an existing factory. Only a derived class has access to the protected member fTransports, in which AddTransport() looks for an existing factory. No test or example uses this, and RESETTING DEVICE clears the member. For channels that share one factory, use channels without a device. See Use channels without a device.

What this means for each transport:

zeromq: Each factory has its own ZeroMQ context, with its own I/O threads (zeromq/TransportFactory.h). An inproc:// address connects only sockets of one context (ZeroMQ behavior). Thus use ipc:// or tcp:// between two devices of one process. The test above uses ipc://.
shmem: Each factory opens the segments of its session and increments the device counter. Devices in one process with the same session (the property session) exchange messages in the same way as devices in different processes. The test ShmMessages.ZeroCopyFromUnmanaged (test/message/_message.cxx) makes two shmem factories with one session in one process and sends messages between them.

The limits are in the parts of FairMQ that exist once per process:

The logger is global. FairLogger keeps its settings in static members. A change of the severity, the verbosity or the log file applies to all devices of the process. The prefix of a log line does not contain the device ID. See Logging.
The signal handling of the control plugin is global. The control plugin installs its handlers for SIGINT and SIGTERM with signal(). It counts the signals in global variables, and no code resets them (Control.cxx). One signal starts the shutdown of each device with a control plugin in the process. The next SIGINT ends the process with std::abort(). A control plugin that starts after a signal, with --catch-signals 1, shuts down its device at once.
The interactive mode reads the standard input (Control.cxx). Two interactive controllers in one process read the same keys.
One DeviceRunner for each process. DeviceRunner sets the global logger, and it loads the control plugin. FairMQ has no test with two DeviceRunner objects in one process. Run the other devices without DeviceRunner, and control them from your code, as the test above does.

For the threads of each device, see Threads in a device.

Read and change properties at run time

The device, the plugins and the hooks of DeviceRunner share one fair::mq::ProgOptions object. Each option of the command line is a property in it. See Configuration and properties.

Code	Access to the properties
A plugin	The property functions of `Plugin`, for example `GetProperty<T>()` and `SetProperty<T>()`.
A device	`GetConfig()` returns a pointer to the `ProgOptions` object.
A hook of `DeviceRunner`	The member `fConfig` of the runner.
A program without `DeviceRunner`	Your `ProgOptions` object, or `GetConfig()` of the device.

Read and set a property

GetProperty<T>(key) throws fair::mq::PropertyNotFoundError when the key does not exist. The text is Config has no key: <key>.
GetProperty<T>(key, fallback) returns fallback when the key does not exist.
T must be the type of the stored value, for example float for rate and int for io-threads. For another type, both functions throw an exception.
GetPropertyAsString(key) returns the value as text. For a type that PropertyHelper does not know, the text is [unidentified_type].
SetProperty<T>(key, value) adds or changes a property. UpdateProperty<T>() changes only a property that exists. It returns false otherwise.
The type of the value is the type of the property. SetProperty("key", "text") stores a const char*. Use SetProperty<std::string>("key", "text") for a text property.

Follow the changes

Subscribe<T>(subscriber, callback) calls callback with the key and the new value of each change. A plugin uses SubscribeToPropertyChange<T>().

The callback runs in the thread that changes the property, before SetProperty() returns (Threads in a device). Protect the data that the callback changes, for example with std::atomic.
The callback gets only the changes with a value of the type T. A subscription for std::string gets no call for a const char* value.
SubscribeAsString() gets the changes of all types, with the value as text.
SetProperty(), UpdateProperty(), SetProperties() and UpdateProperties() call the subscribers. DeleteProperty() and AddChannel() do not.
A subscriber name has one subscription for each type. A second subscription with the same name and type replaces the first.
Unsubscribe<T>(subscriber) throws std::out_of_range when the subscription does not exist.

The dispatcher of the qc example subscribes in InitTask() and unsubscribes in ResetTask(). The value active of the property qc sets the flag fDoQC. The value inactive clears it.

void InitTask() override
{
    GetConfig()->Subscribe<std::string>("qcdevice", [&](const std::string& key, std::string value) {
        if (key == "qc") {
            if (value == "active") {
                fDoQC.store(true);
            } else if (value == "inactive") {
                fDoQC.store(false);
            }
        }
    });
}

examples/qc/qCDispatcher.cxx:20–31

A plugin can set the flag with SetProperty<std::string>("qc", "active"). The callback runs in the thread that calls SetProperty(). Thus fDoQC is a std::atomic<bool>.

When a change has an effect

Property	The device uses a new value
`severity`, `file-severity`, `verbosity`, `log-to-file` (type `std::string`), `color` (type `bool`)	At once. `DeviceRunner` subscribes to these keys after it parses the command line. See Logging.
`id`, `rate`, `init-timeout`, `transport`, `network-interface`, `chans.<name>.<index>.<key>`	In `INITIALIZING DEVICE`, after the request of `CompleteInit`.
`io-threads`, `session`, the options of shared memory	In `INITIALIZING DEVICE`, when the device creates the transport.
`chans.<name>.<index>.address` of a channel that is not connected yet	In `CONNECTING`. The device reads the address again every 50 ms, until the time of `init-timeout` ends.
Your own properties	When your code reads them, or in your subscription callback.

When the device reads a property has the details for each property.

In its state callback of INITIALIZING DEVICE, the config plugin writes the channel properties of --mq-config or --channel-config. The device reads the properties after all state callbacks return and after the request of CompleteInit.
The state callbacks of the plugins run in the sequence in which the plugin manager creates the plugins. The plugins of -P come first, then config, the plugins of the hook LoadPlugins, and control.
After BINDING, chans.<name>.<index>.address holds the address that the channel bound to.
To use new values of the INITIALIZING DEVICE properties, request ResetDevice and initialize the device again. RESETTING DEVICE removes the channels and the transports.

Problems

Problem	Cause and solution
`ChangeState()` or `ChangeDeviceState()` returns `false`.	The log shows the cause. `No transition from state … on transition …`: the current state does not accept the transition. `Transition … incoming, but another state transition is already ongoing.`: wait for the next state first.
The device stays in `INITIALIZING DEVICE`.	The device waits for `CompleteInit`. Request it.
`WaitForState()` does not return.	The state does not come. Check the transition that you requested, and check the log for an exception in a hook.
`DeviceRunner::Run()` does not return after `EXITING`.	A plugin has the controller role. Call `ReleaseDeviceControl()` in `EXITING`.
`RUNNING` ends at once.	The controller requested `Stop` in the callback of `RUNNING`, or the work of the device ended.
A subscription callback gets no call.	The type of the new value is not the type of the subscription. Give the type to `SetProperty<T>()`.
`could not connect all channels after 120 attempts`	A channel did not connect in the time of `init-timeout`. The number in the text is the value of `init-timeout` in seconds.

Plugins and control: the control plugin, its modes and the controller role.
Write a plugin: build and load your own plugin.
custom-controller: the complete controller example.
Device states: the states, the hooks and the state ERROR.