Guides

Receive data with callbacks or loops

This guide shows the three ways to do the work of a device in the state RUNNING: data callbacks, ConditionalRun() and Run(). When you complete it, your device processes messages and leaves RUNNING when its work ends or when a controller requests Stop.

Choose a way

	`OnData()` callbacks	`ConditionalRun()`	`Run()`
Who receives the messages	The device	Your code	Your code
How often the device calls your code	Once for each message	Again and again	Once
How your code ends `RUNNING`	The callback returns `false`	The function returns `false`	The function returns
What happens on `Stop`	The device does not call the callbacks again	The device does not call the function again	Your loop must check `NewStatePending()`
Rate limit	No	The option `--rate`	Your code
Typical device	Sink, processor	Sampler, client of a request and reply	Device with a poller or with its own order of transfers

Use OnData() when your device reacts to messages. Use ConditionalRun() when your device creates data or does a sequence of transfers. Use Run() only when you need full control of the loop.

Receive with OnData()

In the constructor of your device, call OnData() with the name of the input channel and your callback.
Write the callback. It receives the message and the index of the sub-channel.
Return true from the callback to receive the next message. Return false to end RUNNING.

The processor of the example 1-n-1 receives a message on the channel data1 and sends a new message on the channel data2:

struct Processor : fair::mq::Device
{
    Processor()
    {
        OnData("data1", &Processor::HandleData);
    }

    bool HandleData(fair::mq::MessagePtr& msg, int)
    {
        LOG(info) << "Received data, processing...";

        // Modify the received string
        std::string* text = new std::string(static_cast<char*>(msg->GetData()), msg->GetSize());
        *text += " (modified by " + fId + ")";

        // 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 msg2(NewMessage(const_cast<char*>(text->c_str()),
                                        text->length(),
                                        [](void* /*data*/, void* object) { delete static_cast<std::string*>(object); },
                                        text));

        // Send out the output message
        if (Send(msg2, "data2") < 0) {
            return false;
        }

        return true;
    }
};

examples/1-n-1/processor.cxx:16–47

The message belongs to the callback. You can change it, send it, or keep it.

Callback forms

OnData() has four overloads. Two take a member function, and two take a std::function:

Overload	Signature of the callback	Receives
`OnData(name, &T::f)`	`bool f(fair::mq::MessagePtr& msg, int index)`	One message
`OnData(name, callback)`	`InputMsgCallback`	One message
`OnData(name, &T::f)`	`bool f(fair::mq::Parts& parts, int index)`	One multipart message
`OnData(name, callback)`	`InputMultipartCallback`	One multipart message

A multipart callback receives all parts of a message that the sender sent as Parts. This is the start of the callback of the sink in the example multipart. It checks the number of parts and reads the header from the first part:

bool HandleData(fair::mq::Parts& parts, int)
{
    LOG(info) << "Received message with " << parts.Size() << " parts";

    if (parts.Size() != 7) {
        throw std::runtime_error("Number of received parts != 7");
    }

    example_multipart::Header header;
    header.stopFlag = (static_cast<example_multipart::Header*>(parts.At(0)->GetData()))->stopFlag;

    LOG(info) << "Received part 1 (header) with stopFlag: " << header.stopFlag;

examples/multipart/sink.cxx:23–34

A lambda is also a callback. The device of the example qc leaves RUNNING after the first message:

QCTask()
{
    OnData("qc", [](fair::mq::MessagePtr& /*msg*/, int) {
        LOG(info) << "received data";
        return false;
    });
}

examples/qc/qCTask.cxx:15–21

Several input channels

Call OnData() once for each input channel. The sink of the example multiple-channels has two callbacks:

Sink()
{
    OnData("broadcast", &Sink::HandleBroadcast);
    OnData("data", &Sink::HandleData);
}

examples/multiple-channels/sink.cxx:19–23

The device selects the method from the input channels:

One channel with one sub-channel: the device calls Receive() in a loop, without a poller.
More channels or sub-channels, one transport: the device polls all input channels with a timeout of 200 ms. Then it calls the callbacks of the channels with data, in the order of the OnData() calls.
More than one transport: the device polls each transport in its own thread, with a timeout of 500 ms. A mutex makes sure that only one callback runs at a time.

The index argument tells the callback which sub-channel received the message.

Rules for callbacks

Register the callbacks before RUNNING starts, for example in the constructor.
Register one callback for each channel. A second OnData() of the same form for the same channel has no effect.
With callbacks, the device does not call ConditionalRun() and Run(). It calls PreRun() and PostRun().
The device checks for a pending transition between two callbacks. A callback that does not return delays the transition.
If Receive() returns a negative transfer code, the device does not call the callback and ends RUNNING.

Use ConditionalRun()

Override ConditionalRun(). Do one step of the work in each call.
Check the return value of each Send() and Receive(). Return false when it is negative.
Return true to get the next call. Return false to end RUNNING.
To limit the number of calls per second, start the device with the option --rate.

The client of the example req-rep sends a request and waits for the reply in each call:

bool ConditionalRun() override
{

    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 req(NewMessage(const_cast<char*>(text->c_str()), // data
                                                        text->length(), // size
                                                        [](void* /*data*/, void* object) { delete static_cast<std::string*>(object); }, // deletion callback
                                                        text)); // object that manages the data
    fair::mq::MessagePtr rep(NewMessage());

    LOG(info) << "Sending \"" << fText << "\" to server.";

    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;
        }
    }

    return false;
}

examples/req-rep/client.cxx:26–59

The device calls ConditionalRun() while it returns true and no transition is pending. After the last call, the device calls Run() once.

The option --rate gives the maximum rate of the calls in Hz. The default is 0, which means no limit. The device reads the value in INITIALIZING DEVICE. A tools::RateLimiter measures the time between the calls and sleeps when the calls come too fast. Values of 0.001 or less give no limit. To limit a loop of your own, see RateLimit.h.

fairmq-ex-1-1-sampler --id sampler1 --rate 10 --channel-config name=data,type=push,method=bind,address=tcp://*:5555

Write a Run() loop

Override Run().
Write a loop that continues while NewStatePending() returns false.
In the loop, check the return value of each Send() and Receive().
For a pause, call WaitFor() and not std::this_thread::sleep_for(). A pending transition ends WaitFor() at once.
Return from Run() to end RUNNING.

The sender of the example n-m receives an id on the channel sync and sends a multipart message to one of the receivers:

void Run() override
{
    fair::mq::Channel& dataInChannel = GetChannel("sync", 0);

    while (!NewStatePending()) {
        Header h;
        fair::mq::MessagePtr id(NewMessage());
        if (dataInChannel.Receive(id) > 0) {
            h.id = *(static_cast<uint16_t*>(id->GetData()));
            h.senderIndex = fIndex;
        } else {
            continue;
        }

        fair::mq::Parts parts;
        parts.AddPart(NewSimpleMessage(h));
        parts.AddPart(NewMessage(fSubtimeframeSize));

        uint64_t currentDataId = h.id;
        int direction = currentDataId % fNumReceivers;

        if (Send(parts, "data", direction, 0) < 0) {
            LOG(debug) << "Failed to queue Subtimeframe #" << currentDataId << " to Receiver[" << direction << "]";
        }
    }
}

examples/n-m/sender.cxx:29–54

When a controller requests Stop, the device interrupts its transports. The waiting Receive() returns TransferCode::interrupted. The loop continues, and NewStatePending() ends it.

For a loop that also waits for output channels, see Poll several channels.

What the tests show

The tests in test/device/_data_callbacks.cxx check these properties. The DataCallbacks tests run for each transport:

Test	Result
`DataCallbacks.SingleMessage`	The callback receives three messages with the index 0. It returns `false` after the third, and the device goes to `READY`.
`DataCallbacks.Multipart`	The multipart callback receives the parts `a` and `bc` in one call.
`DataCallbacks.TwoChannels`	A message callback and a multipart callback on two channels receive their messages.
`MultipleTransports.OneInputEach`	A zeromq channel and a shmem channel in one device receive their messages.
`ConditionalRun.EndsRunWhenItReturnsFalse`	`ConditionalRun()` returns `false` in the fifth call. The device makes no more calls and goes to `READY`.
`ConditionalRun.Rate`	With `rate` 100, the 21 calls take 140 ms or more.

Check the result

Start the sampler and the sink of the example 1-1 in two terminals:

fairmq-ex-1-1-sampler --id sampler1 --rate 1 --max-iterations 3 --control static --channel-config name=data,type=push,method=bind,address=tcp://*:5555,rateLogging=0

fairmq-ex-1-1-sink --id sink1 --max-iterations 3 --control static --channel-config name=data,type=pull,method=connect,address=tcp://localhost:5555,rateLogging=0

The log of the sink shows these lines (without the prefix of the logger):

READY ---> RUNNING
fair::mq::Device running...
Received: "Hello"
Received: "Hello"
Received: "Hello"
Configured maximum number of iterations reached. Leaving RUNNING state.
RUNNING ---> READY

The line RUNNING ---> READY tells you that the device left RUNNING. With --control static, the device then resets and exits.

Problems that can occur

The device goes to ERROR at the start of RUNNING: the channel name in OnData() does not exist. The device throws std::out_of_range. Make sure that the name is the same as in the channel configuration.
RUNNING ends without a reason in the log: a Receive() for a callback returned a negative value. For example, the input channel has the property rcvTimeoutMs, and no message arrived in time. Remove the timeout, or use a Run() loop.
The device does not leave RUNNING: your Run() loop does not check NewStatePending(), or a callback does not return. Add the check, and return from the callbacks. A second Ctrl-C aborts a device that does not stop.
ConditionalRun() is not called: the device has data callbacks. Remove the OnData() calls, or move the work into the callbacks.
A multipart callback receives wrong parts: the sender sends single messages. With shmem, a multipart Receive() of a single message reads wrong metadata. See the disabled test PartsTransfer.SingleSendMultipartReceive. Use the same form on both sides.

Device states: what happens in RUNNING, and what ends it.
Devices: the hooks and the helper functions of a device.
Handle errors and timeouts: transfer codes, timeouts and exceptions.
Poll several channels: a Run() loop with a poller.
Messages and parts: MessagePtr and Parts.