Devices

A device is the part of a FairMQ topology that does the work. You write a class derived from fair::mq::Device and override its hooks. The state machine of the device calls the hooks, and the channels of the device connect it to other devices.

What a device has

Each device has these parts:

Part	Access	Description
Channels	GetChannels(), GetChannel()	A map from the channel name to a std::vector<Channel> with the sub-channels. The device makes the channels from its properties in INITIALIZING DEVICE.
Configuration	fConfig, GetConfig()	A pointer to a ProgOptions object. It holds all properties of the device, also the channel properties.
Transports	Transport(), AddTransport()	One transport factory for each transport that the device uses. Transport() returns the default transport.
State machine	ChangeState(), GetCurrentState()	A private member. It calls the hooks. See Device states.
Id	GetId()	The value of the property id. The device reads it in INITIALIZING DEVICE.
Version	GetVersion()	The tools::Version that you give to the constructor. The default is {0, 0, 0}.

The public member fChannels is deprecated. Use GetChannels().

A device has four constructors. Copy and move are deleted.

fairmq/Device.h

Device();
Device(ProgOptions& config);
Device(tools::Version version);
Device(ProgOptions& config, tools::Version version);

fairmq/Device.h:60–63

If you do not give a ProgOptions object, the device makes its own (fInternalConfig). SetConfig() replaces it with your object. DeviceRunner calls SetConfig() with its own ProgOptions before the device starts.

The transports

In INITIALIZING DEVICE, the device reads the property transport (default zeromq). It makes the default transport from this value. Then it makes one more transport for each other transport that a channel uses. All channels with the same transport share one transport factory. See Transports and Use several transports.

The hooks

The hooks are protected virtual functions. Their default implementations do nothing, and ConditionalRun() returns false. Override only the hooks that your device needs.

Hook	State	When the device calls it
Init()	INITIALIZING DEVICE	After the controller requests CompleteInit, before the device makes the channels. The header recommends InitTask() instead.
Bind()	BINDING	After the device binds all channels with the method bind.
Connect()	CONNECTING	After the device connects all channels with the method connect.
InitTask()	INITIALIZING TASK	Once. Read your options and allocate your resources here.
PreRun()	RUNNING	Once, at the start of RUNNING.
ConditionalRun()	RUNNING	Again and again, while it returns true and no transition is pending.
Run()	RUNNING	Once, after the ConditionalRun() loop.
PostRun()	RUNNING	Once, at the end of RUNNING. A transition is pending at this time.
ResetTask()	RESETTING TASK	Once. Undo the work of InitTask() here.
Reset()	RESETTING DEVICE	After the device resets its transports, before it removes all channels.

The device does not call ConditionalRun() and Run() when it has data callbacks (see below). The state machine calls the hooks through a handler in the constructor of the device:

fairmq/Device.cxx

fStateMachine.HandleStates([&](State state) {
    LOG(trace) << "device notified on new state: " << state;

    fStateQueue.Push(state);

    switch (state) {
        case State::InitializingDevice:
            InitWrapper();
            break;
        case State::Binding:
            BindWrapper();
            break;
        case State::Connecting:
            ConnectWrapper();
            break;
        case State::InitializingTask:
            InitTaskWrapper();
            break;
        case State::Running:
            RunWrapper();
            break;
        case State::ResettingTask:
            ResetTaskWrapper();
            break;
        case State::ResettingDevice:
            ResetWrapper();
            break;
        case State::Exiting:
            Exit();
            break;
        default:
            LOG(trace) << "device notified on new state without a matching handler: " << state;
            break;
    }
});

fairmq/Device.cxx:106–140

Each wrapper function calls one or more hooks. For example, InitTaskWrapper() calls InitTask() and then requests the transition Auto, if no other transition is pending. An exception in a hook moves the device to the state ERROR. See Handle errors and timeouts.

Two more virtual functions are public:

• RegisterChannelEndpoints(): DeviceRunner calls it before the device starts. In it, call RegisterChannelEndpoint() for each channel that your device expects, with the minimum and maximum number of sub-channels (default 1 and 1).
• LogSocketRates(): in RUNNING, a separate thread runs it when a channel has the property rateLogging larger than 0 (default 1). It logs the message rate and the data rate of these channels every rateLogging seconds.

A small example

This sampler of the example 1-1 reads two options in InitTask() and sends one message in each call of ConditionalRun():

examples/1-1/sampler.cxx

struct Sampler : fair::mq::Device
{
    void InitTask() override
    {
        // Get the fText and fMaxIterations values from the command line options (via fConfig)
        fText = fConfig->GetProperty<std::string>("text");
        fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations");
    }

    bool ConditionalRun() override
    {
        // create a copy of the data with new(), that will be deleted after the transfer is complete
        std::string* text = new std::string(fText);

        // create message object with a pointer to the data buffer, its size,
        // custom deletion function (called when transfer is done),
        // and pointer to the object managing the data buffer
        fair::mq::MessagePtr msg(NewMessage(
            const_cast<char*>(text->c_str()),
            text->length(),
            [](void* /*data*/, void* object) { delete static_cast<std::string*>(object); },
            text));

        LOG(info) << "Sending \"" << fText << "\"";

        // in case of error or transfer interruption, return false to go to the Ready state
        // successfull transfer will return number of bytes transfered (can be 0 if sending an empty message).
        if (Send(msg, "data") < 0) {
            return false;
        } else if (fMaxIterations > 0 && ++fNumIterations >= fMaxIterations) {
            LOG(info) << "Configured maximum number of iterations reached. Leaving RUNNING state.";
            return false;
        }

        return true;
    }

  private:
    std::string fText;
    uint64_t fMaxIterations = 0;
    uint64_t fNumIterations = 0;
};

examples/1-1/sampler.cxx:16–57

Data callbacks

A device can register data callbacks with OnData(). The device then calls the callback for each message that arrives on the channel. A callback receives a message (MessagePtr&) or a multipart message (Parts&), and the index of the sub-channel. It returns true to continue, or false to end RUNNING.

examples/1-1/sink.cxx

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

examples/1-1/sink.cxx:18–22

Register the callbacks before the device enters RUNNING, for example in the constructor. See Receive data with callbacks or loops for all forms of OnData(). For the thread that calls the callbacks, see Which thread runs your code.

Helper functions

The device has helper functions for the work in the hooks:

Function	Description
Send(msg, channel, index = 0), Receive(msg, channel, index = 0)	Send or receive a MessagePtr, a Parts or a std::vector<MessagePtr> on the sub-channel index of the channel. Overloads with a fourth argument set a timeout in ms (Send-2, Receive-2). They return the number of bytes or a negative transfer code.
NewMessage(args...)	Makes a message with the default transport. The arguments go to TransportFactory::CreateMessage().
NewMessageFor(channel, index, args...)	Makes a message with the transport of the sub-channel.
NewSimpleMessage(data), NewSimpleMessageFor(channel, index, data)	Makes a message with a copy of data.
NewStaticMessage(data), NewStaticMessageFor(channel, index, data)	Makes a message that does not free data after the transfer. The zeromq transport uses data without a copy. The shmem transport copies it.
NewUnmanagedRegion(args...), NewUnmanagedRegionFor(channel, index, args...)	Makes an unmanaged region. See Unmanaged regions.
NewPoller("a", "b"), NewPoller(channels)	Makes a poller for channel names or for a std::vector<Channel*>. All channels must use the same transport. Otherwise the function throws std::runtime_error. See Poll several channels.
GetChannel(name, index = 0)	Returns the sub-channel. Throws std::out_of_range if it does not exist.
GetNumSubChannels(name)	Returns the number of sub-channels of a channel.
GetNumberOfConnectedPeers(name, index = 0)	Returns the number of peers that are connected to the sub-channel.
AddChannel(name, channel)	Adds the properties of a Channel object to the configuration, except sndTimeoutMs and rcvTimeoutMs. The device makes the channel in INITIALIZING DEVICE.
GetConfig()	Returns the ProgOptions object. Use GetConfig()->GetProperty<T>(key) to read an option.
GetId()	Returns the id of the device.
NewStatePending()	Returns true when a transition is requested and the next state has not started yet.
WaitFor(duration)	Waits for the duration. Returns false if a transition is requested before the end of the duration.
ChangeState(transition), ChangeStateOrThrow(transition)	Request a transition. See States and transitions.
GetCurrentState(), GetCurrentStateName()	Return the current state.
WaitForState(state), WaitForNextState()	Wait in another thread for a state of the device.
SubscribeToStateChange(key, callback)	Calls callback at the start of each new state, in the thread of the state machine.
RunStateMachine()	Runs the state machine in the calling thread, until the state EXITING or ERROR.

Some setters change a property in the configuration, for example SetTransport() (transport) and SetNumIoThreads() (io-threads). The device reads most of these properties in INITIALIZING DEVICE. For all properties, see Configuration and properties.

How a device starts

Most devices use the header fairmq/runDevice.h. It defines a main() function that runs one device with the class DeviceRunner.

Use runDevice.h

1. Include <fairmq/runDevice.h> in one source file of your executable.

2. Write the function addCustomOptions(). Add your options to the options_description. If you have no options, leave the body empty.

   examples/1-1/sampler.cxx

   void addCustomOptions(bpo::options_description& options)
   {
       options.add_options()
           ("text", bpo::value<std::string>()->default_value("Hello"), "Text to send out")
           ("max-iterations", bpo::value<uint64_t>()->default_value(0), "Maximum number of iterations of Run/ConditionalRun/OnData (0 - infinite)");
   }

   examples/1-1/sampler.cxx:59–64

3. Write the function getDevice(). Return a new object of your device class.

   examples/1-1/sampler.cxx

   std::unique_ptr<fair::mq::Device> getDevice(fair::mq::ProgOptions& /*config*/)
   {
       return std::make_unique<Sampler>();
   }

   examples/1-1/sampler.cxx:66–69

The main() function of runDevice.h adds these two functions as hooks of DeviceRunner:

fairmq/runDevice.h

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

    try {
        DeviceRunner runner(argc, argv);

        // runner.AddHook<LoadPlugins>([](DeviceRunner& r){
        //     // for example:
        //     r.fPluginManager->SetSearchPaths({"/lib", "/lib/plugins"});
        //     r.fPluginManager->LoadPlugin("asdf");
        // });

        runner.AddHook<SetCustomCmdLineOptions>([](DeviceRunner& r){
            boost::program_options::options_description customOptions("Custom options");
            addCustomOptions(customOptions);
            r.fConfig.AddToCmdLineOptions(customOptions);
        });

        // runner.AddHook<ModifyRawCmdLineArgs>([](DeviceRunner& r){
        //     // for example:
        //     r.fRawCmdLineArgs.push_back("--blubb");
        // });

        runner.AddHook<InstantiateDevice>([](DeviceRunner& r){
            r.fDevice = getDevice(r.fConfig);
        });

        return runner.Run();

        // Run with builtin catch all exception handler, just:
        // return runner.RunWithExceptionHandlers();
    } catch (std::exception& e) {
        LOG(error) << "Uncaught exception reached the top of main: " << e.what();
        return 1;
    } catch (...) {
        LOG(error) << "Uncaught exception reached the top of main.";
        return 1;
    }
}

fairmq/runDevice.h:19–59

An exception that reaches main() gives the log line Uncaught exception reached the top of main: and the exit code 1.

Old header

fairmq/runFairMQDevice.h is the old form of this header. There, getDevice() takes a const fair::mq::ProgOptions& and returns a raw pointer. Use fairmq/runDevice.h for new code.

What DeviceRunner does

DeviceRunner::Run() starts the device in this sequence. The constructor of DeviceRunner already loads the plugins of the option -P.

1. Load the built-in plugin config.
2. Call the hook LoadPlugins.
3. Load the built-in plugin control.
4. Call the hook SetCustomCmdLineOptions.
5. Add the options of all plugins and of the plugin manager.
6. Call the hook ModifyRawCmdLineArgs.
7. Parse the command line. For --help or --print-options, print the output and return 0.
8. Set up the logger from the logging options.
9. Call the hook InstantiateDevice. If the hook does not set fDevice, log an error and return 1.
10. Call RegisterChannelEndpoints() of the device. For --print-channels or --version, return 0.
11. Give the configuration to the device (SetConfig()), and instantiate the plugins.
12. Log the configuration, and call RunStateMachine() of the device.
13. When the state machine stops, wait until all plugins release the device control. Return 0.

You add a hook with AddHook<H>(). A hook is a function void(DeviceRunner&). It can use all members of the runner: fRawCmdLineArgs, fConfig, fDevice and fPluginManager.

Hook (fair::mq::hooks::)	Typical use
LoadPlugins	Load more plugins with r.fPluginManager.LoadPlugin().
SetCustomCmdLineOptions	Add options with r.fConfig.AddToCmdLineOptions().
ModifyRawCmdLineArgs	Change the arguments in r.fRawCmdLineArgs before the parser reads them.
InstantiateDevice	Make the device and put it into r.fDevice.

You can write your own main() with DeviceRunner. The example custom-controller loads its own control plugin in the hook LoadPlugins:

examples/custom-controller/main.cxx

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:17–46

RunWithExceptionHandlers() calls Run() and catches all exceptions. It logs Uncaught exception reached the top of DeviceRunner: and returns 1.

Known defect

For --print-channels and --version, DeviceRunner sets the console severity to nolog and then prints with LOG. Thus both options print nothing in this version. The device does not start.

A process can also run more than one device without DeviceRunner. Then your code must call RunStateMachine() and request the transitions. See Control a device from code.

The life of a device

A device starts in the state IDLE. A controller, usually the control plugin, requests the transitions. The device initializes in steps, for example BINDING and CONNECTING, until it is READY. In RUNNING, the device processes data. When the work ends, the device goes back to READY. Then it resets in two steps back to IDLE, and ends in EXITING.

For the states, the transitions and the hooks in each state, see Device states.

Related pages

• Device states: the state machine and the hooks in each state.
• Receive data with callbacks or loops: OnData(), ConditionalRun() and Run().
• Channels and sockets: the channels of a device.
• Configuration and properties: the properties that a device reads.
• Plugins and control: the plugins that control the device.
• Threads in a device: the thread that runs each hook and callback.