Concepts

Threads in a device

A device process has more than one thread. FairMQ, its plugins and ZeroMQ start threads, and your code can add more, for example in PreRun(). This page tells you which thread runs each hook and callback, which objects are thread-safe, and how to add a thread without a data race.

The threads of a device process

A device that starts with fairmq/runDevice.h can have these threads:

Thread	Started by	When it runs	What it does
Main thread	The process	Always	Runs `main()` and `DeviceRunner::Run()`. It creates the plugins and then calls `RunStateMachine()`. This thread is the thread of the state machine.
Controller thread	The control plugin	With each value of `--control` except `none`	Requests the transitions. In the mode `dynamic` (default), it also reads the keys of the terminal (Control.cxx).
Signal handler thread	The control plugin	With `--catch-signals 1` (default)	Checks every 100 ms for `SIGINT` or `SIGTERM`. After the first signal, it stops the controller thread and requests the transitions of the shutdown sequence (Control.cxx).
Rate logger thread	The device	In `RUNNING`, if a sub-channel has `rateLogging` above 0 (default 1)	Runs `LogSocketRates()`. It ends when a transition is pending. The device joins it after `PostRun()` (Device.cxx).
Poll threads	The device	In `RUNNING`, with data callbacks on input channels of more than one transport	One thread for each transport. Each thread polls its channels and calls their data callbacks (Device.cxx).
ZeroMQ I/O threads	ZeroMQ	While the transport exists	Each transport, also `shmem`, has one ZeroMQ context with `--io-threads` I/O threads (default 1) (zeromq/TransportFactory.h, shmem/TransportFactory.h). ZeroMQ moves the data of the sockets in these threads (ZeroMQ behavior).
Heartbeat thread	The shmem transport	While the transport exists	Writes the heartbeat for `fairmq-shmmonitor` every 100 ms (shmem/Manager.h).
Ack threads of a region	The shmem transport	While the region exists	The ack sender sends the released blocks to the creator of the region. The ack receiver of the creator calls the region callback. See Acknowledgements.
Region event thread	The transport	After `SubscribeToRegionEvents()`	Calls the region event callback. See Region events.

The control plugin does not start a controller thread with --control none, or when another plugin has the controller role. The signal handler thread still runs. The C signal handler itself only counts the signals and stores the last one. A SIGINT after a first signal calls std::abort() there. See Signals.

Without DeviceRunner, the device has no plugins. Then the thread of the state machine is the thread that calls RunStateMachine(), and your code requests the transitions from another thread. See Run the device in a thread. One process can run several devices in this way. In the test MultipleDevices.TwoInSameProcess, two devices run in one process, each with its own thread for RunStateMachine() and its own thread for the transitions. For the limits, see Several devices in one process.

Which thread runs your code

Your code	Thread	Notes
The hooks `Init()`, `Bind()`, `Connect()`, `InitTask()`, `PreRun()`, `ConditionalRun()`, `Run()`, `PostRun()`, `ResetTask()`, `Reset()`	The thread of the state machine	One hook at a time. The next state starts only after the hook returns. See The hooks.
Data callbacks of `OnData()`	The thread of the state machine. With input channels of more than one transport: the poll thread of the transport.	In the poll threads, a mutex lets one callback run at a time (Device.cxx). See Several input channels.
State change callbacks: `SubscribeToStateChange()`, `SubscribeToDeviceStateChange()` of a plugin	The thread of the state machine	At the start of each state, before the hooks of the state (StateMachine.cxx).
New transition callbacks: `SubscribeToNewTransition()`	The thread that calls `ChangeState()`	The state machine holds its lock while it calls the callback (StateMachine.cxx).
Property callbacks: `Subscribe<T>()`, `SubscribeAsString()`, `SubscribeToPropertyChange<T>()`	The thread that changes the property, for example with `SetProperty()`	Before the call returns, after the property store releases its lock (ProgOptions.h).
Region callbacks: `RegionCallback`, `RegionBulkCallback`	shmem: the ack receiver thread of the region. zeromq: the thread that makes the region message.	The zeromq transport copies the data and calls the callback at once, in `NewMessage()` (zeromq/Message.h).
Region event callbacks	The region event thread of the transport	The shmem transport checks for events every 50 ms.
Message free functions (`FreeFn`)	shmem: the thread that makes the message. zeromq: a thread that ZeroMQ selects.	The shmem transport copies the buffer and calls the function at once. ZeroMQ calls it when it releases the buffer, also from its own threads (ZeroMQ behavior). See Your own buffer.
Your override of `LogSocketRates()`	The rate logger thread	Only in `RUNNING`.
Constructors of plugins, hooks of `DeviceRunner`	The main thread	Before the state machine starts.

Some rules follow from this table:

A state change callback blocks the device. The device calls the hooks of the state only after all state callbacks return. For long work, start a thread. See A controller with its own thread.
Do not call ChangeState(), ChangeStateOrThrow() or WaitFor() in a new transition callback. These functions lock the mutex that the state machine holds during the callback.
The device itself subscribes to new transitions. Its callback interrupts the transports, in the thread that requests the transition. The next state resumes the transports in the thread of the state machine (Device.cxx).
A zeromq channel that sends a shmem message wraps the message. The free function of the wrapper destroys the shmem message (Channel.h). Thus the shmem message can be destroyed in a thread of ZeroMQ.
Both transports call the region event callback while they hold the lock of the subscription (shmem/Manager.h, zeromq/Context.h). Do not call SubscribeToRegionEvents() or UnsubscribeFromRegionEvents() in the callback. With zeromq, also do not create or destroy a region and do not call GetRegionInfo() there. These functions lock the same mutex.

What is thread-safe

Object or function	From several threads at the same time	Source
`Channel` and its socket	No. The header says: "The class is not thread-safe." Use each channel from one thread at a time.	Channel.h
Different channels, also of one transport	Yes. Each thread uses its own channels.	Test `PushPullMultipart.MultiThreaded_ipc`
`GetNumberOfConnectedPeers()` of a channel	Safe for distinct channels, not safe for one shared channel.	Channel.h
`Message`, `Parts`, a `Poller`	No. These classes have no lock. Give each object to one thread at a time.	Message.h, Parts.h, Poller.h
`ChannelResource`	No. Its map of the allocated messages has no lock. See Use std::pmr memory resources.	MemoryResources.h
The property functions of `ProgOptions`, for example `GetProperty<T>()` and `SetProperty<T>()`	Yes. Each function locks a mutex. `GetVarMap()` returns a reference, and the lock does not protect it.	ProgOptions.h
`ChangeState()`, `NewStatePending()`, `WaitFor()`, `GetCurrentState()`	Yes. The state machine uses a mutex and atomic values.	StateMachine.cxx, StateMachine.cxx
Subscribe and unsubscribe for state changes	Yes. A mutex protects the subscriptions.	StateMachine.cxx, test `PluginServices.SubscriptionThreadSafety`
`WaitForState()`, `WaitForNextState()`	Call them from one thread only. The queue of states gives each state to one caller.	StateQueue.h
`Interrupt()`, `Resume()` of a transport	Yes. They set an atomic flag.	zeromq/Context.h, shmem/Manager.h, test `InterruptTransfer.Receive`
The byte and message counters of a channel, for example `GetBytesTx()`	Yes. The sockets keep them in atomic values. The rate logger thread reads them.	zeromq/Socket.h, shmem/Socket.h

The device makes its channels in INITIALIZING DEVICE, after Init() (Device.cxx). It removes them in RESETTING DEVICE (Device.cxx). Between these states, the map of the channels does not change. Thus several threads can call GetChannel(), Send() and Receive() of the device, each for its own channels. Register the data callbacks before RUNNING, and add the channels in Init() at the latest. Do both in one thread.

The test PushPullMultipart.MultiThreaded_ipc binds and connects two channels of one transport in the main thread. Then it sends on one channel and receives on the other, each in its own thread:

auto RunMultiThreadedMultipart(test::TransportParam const& param, string const& address1) -> void
{
    ProgOptions config;
    test::Configure(config, tools::Uuid(), param.expandedShmMetadata);
    config.SetProperty<int>("io-threads", 1);
    config.SetProperty<size_t>("shm-segment-size", 20000000); // NOLINT

    auto factory = TransportFactory::CreateTransportFactory(param.transport, tools::Uuid(), &config);

    Channel push1("Push1", "push", factory);
    ASSERT_TRUE(push1.Bind(address1));
    Channel pull1("Pull1", "pull", factory);
    ASSERT_TRUE(pull1.Connect(address1));

    auto pusher = thread{[&push1](){
        ASSERT_TRUE(push1.Validate());

        Parts sent;
        sent.AddPart(push1.NewSimpleMessage("1"));
        sent.AddPart(push1.NewSimpleMessage("2"));
        sent.AddPart(push1.NewSimpleMessage("3"));

        ASSERT_GE(push1.Send(sent), 0);
    }};

    auto puller = thread{[&pull1](){
        ASSERT_TRUE(pull1.Validate());

        Parts received;
        ASSERT_GE(pull1.Receive(received), 0);

        stringstream out;
        for_each(received.cbegin(), received.cend(), [&out](const MessagePtr& part) {
            out << string{static_cast<char*>(part->GetData()), part->GetSize()};
        });
        ASSERT_EQ(out.str(), "123");
    }};

    pusher.join();
    puller.join();
}

test/protocols/_push_pull_multipart.cxx:107–147

Safe patterns

One thread for each channel

Start your thread in PreRun(), and join it in PostRun(). Give the thread its own channels, and do not use these channels in the other threads. The sampler of the example multiple-transports sends on data1 in ConditionalRun(). A second thread receives the acknowledgements on ack:

void PreRun() override
{
    fAckListener = std::thread(&Sampler1::ListenForAcks, this);
}

examples/multiple-transports/sampler1.cxx:23–26

void PostRun() override
{
    fAckListener.join();
}

examples/multiple-transports/sampler1.cxx:47–50

void ListenForAcks()
{
    uint64_t numAcks = 0;

    while (!NewStatePending()) {
        fair::mq::MessagePtr ack(NewMessageFor("ack", 0));
        if (Receive(ack, "ack") < 0) {
            break;
        }
        ++numAcks;
    }

    LOG(info) << "Acknowledged " << numAcks << " messages";
}

examples/multiple-transports/sampler1.cxx:52–65

This pattern works for these reasons:

PostRun() runs while the transition is pending. The device interrupted the transports, so the waiting Receive() returns TransferCode::interrupted, and the loop ends.
The next state resumes the transports only after PostRun() returns. Thus the thread cannot wait again before the join.
The channels exist until RESETTING DEVICE. The thread ends before this state.

Stop your thread with the state machine

Check NewStatePending() in the loop of your thread.
For a pause, call WaitFor() and not std::this_thread::sleep_for(). WaitFor() returns false at once when a transition is requested. See Wait for a transition in your code.
Check the return value of each Send() and Receive(). A negative value ends the loop.
Catch the exceptions in the function of your thread. An exception that leaves the function of a std::thread calls std::terminate(), and the device does not go to ERROR. To report the error, request the transition ErrorFound, for example with ChangeStateOrThrow(). The state machine accepts it in all states.

Hand messages over by move

A MessagePtr is a std::unique_ptr<Message>. To give a message to another thread, move it into a queue that a mutex protects. Then only one thread owns the message at a time. In this device, the data callback puts each message into a queue. A second thread takes the messages out and sends them on its own channel out:

#include <fairmq/Device.h>

#include <chrono>
#include <condition_variable>
#include <deque>
#include <mutex>
#include <thread>

struct Forwarder : fair::mq::Device
{
    Forwarder() { OnData("in", &Forwarder::HandleData); }

    bool HandleData(fair::mq::MessagePtr& msg, int /*index*/)
    {
        {
            std::lock_guard<std::mutex> lock(fMtx);
            fQueue.push_back(std::move(msg));
        }
        fCV.notify_one();
        return true;
    }

    void PreRun() override { fSender = std::thread(&Forwarder::SendLoop, this); }

    void PostRun() override
    {
        fSender.join();
        fQueue.clear(); // destroy the messages that were not sent
    }

    void SendLoop()
    {
        while (!NewStatePending()) {
            fair::mq::MessagePtr msg;
            {
                std::unique_lock<std::mutex> lock(fMtx);
                if (!fCV.wait_for(lock, std::chrono::milliseconds(100), [this] { return !fQueue.empty(); })) {
                    continue;
                }
                msg = std::move(fQueue.front());
                fQueue.pop_front();
            }
            if (Send(msg, "out") < 0) {
                break;
            }
        }
    }

    std::mutex fMtx;
    std::condition_variable fCV;
    std::deque<fair::mq::MessagePtr> fQueue;
    std::thread fSender;
};

The callback uses the channel in, in the thread of the state machine. The second thread uses the channel out. No channel is in two threads.
After the move, the callback does not use msg again.
PostRun() destroys the messages that remain in the queue. Destroy all messages before RESETTING DEVICE, because the device then destroys its transports.

Protect the data of callbacks

A region callback, a property callback and the data callbacks of several transports can run in other threads than your hooks. Protect the data that they share with a mutex or a std::atomic. The readout device of the example readout counts the acknowledgements in a std::atomic<uint64_t>:

void InitTask() override
{
    fMsgSize = fConfig->GetProperty<int>("msg-size");
    fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations");

    fRegion = fair::mq::UnmanagedRegionPtr(NewUnmanagedRegionFor("rb",
                                                            0,
                                                            10000000,
                                                            [this](const std::vector<fair::mq::RegionBlock>& blocks) { // callback to be called when message buffers no longer needed by transport
                                                                fNumUnackedMsgs -= blocks.size();
                                                                if (fMaxIterations > 0) {
                                                                    LOG(debug) << "Received " << blocks.size() << " acks";
                                                                }
                                                            }
                                                            ));
}

examples/readout/readout.cxx:20–35

The member is std::atomic<uint64_t> fNumUnackedMsgs. ConditionalRun() increments it after each successful Send(), and the callback decrements it in the ack receiver thread.

The dispatcher of the example qc uses a std::atomic<bool> for a flag that a property callback sets. See Follow the changes.

Problems that can occur

The device does not leave RUNNING: your thread waits and PostRun() waits in join(). Make the thread check NewStatePending(), and use WaitFor() for pauses.
The process ends with std::terminate(): an exception left the function of a thread, or a data callback threw in a poll thread. Catch the exceptions in the thread. See Receive from channels of different transports.
A crash or wrong data in a transfer: two threads use one channel. Give each channel to one thread.
WaitForState() does not return: another thread took the state from the queue with WaitForState() or WaitForNextState(). Wait for the states in one thread only.

Device states: the state machine and the hooks in each state.
Receive data with callbacks or loops: data callbacks and the poll threads.
Use several transports: channels of two transports in one device.
Control a device from code: controller threads and callbacks.
Unmanaged regions: the ack threads and the region callbacks.