FairMQDevice.h
/********************************************************************************
* Copyright (C) 2012-2018 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#ifndef FAIRMQDEVICE_H_
#define FAIRMQDEVICE_H_
#include <FairMQStateMachine.h>
#include <FairMQTransportFactory.h>
#include <fairmq/Transports.h>
#include <FairMQSocket.h>
#include <FairMQChannel.h>
#include <FairMQMessage.h>
#include <FairMQParts.h>
#include <FairMQUnmanagedRegion.h>
#include <FairMQLogger.h>
#include <options/FairMQProgOptions.h>
#include <vector>
#include <memory> // unique_ptr
#include <algorithm> // std::sort()
#include <string>
#include <chrono>
#include <iostream>
#include <unordered_map>
#include <functional>
#include <assert.h> // static_assert
#include <type_traits> // is_trivially_copyable
#include <stdexcept>
#include <mutex>
#include <condition_variable>
#include <fairmq/Tools.h>
using FairMQChannelMap = std::unordered_map<std::string, std::vector<FairMQChannel>>;
using InputMsgCallback = std::function<bool(FairMQMessagePtr&, int)>;
using InputMultipartCallback = std::function<bool(FairMQParts&, int)>;
class FairMQDevice : public FairMQStateMachine
{
friend class FairMQChannel;
public:
FairMQDevice();
FairMQDevice(FairMQProgOptions& config);
FairMQDevice(const fair::mq::tools::Version version);
FairMQDevice(FairMQProgOptions& config, const fair::mq::tools::Version version);
private:
FairMQDevice(FairMQProgOptions* config, const fair::mq::tools::Version version);
public:
FairMQDevice(const FairMQDevice&) = delete;
FairMQDevice operator=(const FairMQDevice&) = delete;
virtual ~FairMQDevice();
void CatchSignals();
virtual void LogSocketRates();
void SortChannel(const std::string& name, const bool reindex = true);
template<typename Serializer, typename DataType, typename... Args>
void Serialize(FairMQMessage& msg, DataType&& data, Args&&... args) const
{
Serializer().Serialize(msg, std::forward<DataType>(data), std::forward<Args>(args)...);
}
template<typename Deserializer, typename DataType, typename... Args>
void Deserialize(FairMQMessage& msg, DataType&& data, Args&&... args) const
{
Deserializer().Deserialize(msg, std::forward<DataType>(data), std::forward<Args>(args)...);
}
int Send(FairMQMessagePtr& msg, const std::string& channel, const int index = 0, int sndTimeoutInMs = -1)
{
return GetChannel(channel, index).Send(msg, sndTimeoutInMs);
}
int Receive(FairMQMessagePtr& msg, const std::string& channel, const int index = 0, int rcvTimeoutInMs = -1)
{
return GetChannel(channel, index).Receive(msg, rcvTimeoutInMs);
}
int SendAsync(FairMQMessagePtr& msg, const std::string& channel, const int index = 0) __attribute__((deprecated("For non-blocking Send, use timeout version with timeout of 0: Send(msg, \"channelA\", subchannelIndex, timeout);")))
{
return GetChannel(channel, index).Send(msg, 0);
}
int ReceiveAsync(FairMQMessagePtr& msg, const std::string& channel, const int index = 0) __attribute__((deprecated("For non-blocking Receive, use timeout version with timeout of 0: Receive(msg, \"channelA\", subchannelIndex, timeout);")))
{
return GetChannel(channel, index).Receive(msg, 0);
}
int64_t Send(FairMQParts& parts, const std::string& channel, const int index = 0, int sndTimeoutInMs = -1)
{
return GetChannel(channel, index).Send(parts.fParts, sndTimeoutInMs);
}
int64_t Receive(FairMQParts& parts, const std::string& channel, const int index = 0, int rcvTimeoutInMs = -1)
{
return GetChannel(channel, index).Receive(parts.fParts, rcvTimeoutInMs);
}
int64_t SendAsync(FairMQParts& parts, const std::string& channel, const int index = 0) __attribute__((deprecated("For non-blocking Send, use timeout version with timeout of 0: Send(parts, \"channelA\", subchannelIndex, timeout);")))
{
return GetChannel(channel, index).Send(parts.fParts, 0);
}
int64_t ReceiveAsync(FairMQParts& parts, const std::string& channel, const int index = 0) __attribute__((deprecated("For non-blocking Receive, use timeout version with timeout of 0: Receive(parts, \"channelA\", subchannelIndex, timeout);")))
{
return GetChannel(channel, index).Receive(parts.fParts, 0);
}
auto Transport() const -> FairMQTransportFactory*
{
return fTransportFactory.get();
}
template<typename... Args>
FairMQMessagePtr NewMessage(Args&&... args)
{
return Transport()->CreateMessage(std::forward<Args>(args)...);
}
template<typename... Args>
FairMQMessagePtr NewMessageFor(const std::string& channel, int index, Args&&... args)
{
return GetChannel(channel, index).NewMessage(std::forward<Args>(args)...);
}
template<typename T>
FairMQMessagePtr NewStaticMessage(const T& data)
{
return Transport()->NewStaticMessage(data);
}
template<typename T>
FairMQMessagePtr NewStaticMessageFor(const std::string& channel, int index, const T& data)
{
return GetChannel(channel, index).NewStaticMessage(data);
}
template<typename T>
FairMQMessagePtr NewSimpleMessage(const T& data)
{
return Transport()->NewSimpleMessage(data);
}
template<typename T>
FairMQMessagePtr NewSimpleMessageFor(const std::string& channel, int index, const T& data)
{
return GetChannel(channel, index).NewSimpleMessage(data);
}
FairMQUnmanagedRegionPtr NewUnmanagedRegion(const size_t size)
{
return Transport()->CreateUnmanagedRegion(size);
}
FairMQUnmanagedRegionPtr NewUnmanagedRegionFor(const std::string& channel, int index, const size_t size, FairMQRegionCallback callback = nullptr)
{
return GetChannel(channel, index).Transport()->CreateUnmanagedRegion(size, callback);
}
template<typename ...Ts>
FairMQPollerPtr NewPoller(const Ts&... inputs)
{
std::vector<std::string> chans{inputs...};
// if more than one channel provided, check compatibility
if (chans.size() > 1)
{
fair::mq::Transport type = GetChannel(chans.at(0), 0).Transport()->GetType();
for (unsigned int i = 1; i < chans.size(); ++i)
{
if (type != GetChannel(chans.at(i), 0).Transport()->GetType())
{
LOG(error) << "poller failed: different transports within same poller are not yet supported. Going to ERROR state.";
throw std::runtime_error("poller failed: different transports within same poller are not yet supported.");
}
}
}
return GetChannel(chans.at(0), 0).Transport()->CreatePoller(fChannels, chans);
}
FairMQPollerPtr NewPoller(const std::vector<FairMQChannel*>& channels)
{
// if more than one channel provided, check compatibility
if (channels.size() > 1)
{
fair::mq::Transport type = channels.at(0)->Transport()->GetType();
for (unsigned int i = 1; i < channels.size(); ++i)
{
if (type != channels.at(i)->Transport()->GetType())
{
LOG(error) << "poller failed: different transports within same poller are not yet supported. Going to ERROR state.";
throw std::runtime_error("poller failed: different transports within same poller are not yet supported.");
}
}
}
return channels.at(0)->Transport()->CreatePoller(channels);
}
void WaitForInitialValidation() __attribute__((deprecated("This method will have no effect in future versions and will be removed. Instead subscribe for state changes and inspect configuration values."))) {}
std::shared_ptr<FairMQTransportFactory> AddTransport(const fair::mq::Transport transport);
void SetConfig(FairMQProgOptions& config);
FairMQProgOptions* GetConfig() const
{
return fConfig;
}
static bool SortSocketsByAddress(const FairMQChannel &lhs, const FairMQChannel &rhs);
// overload to easily bind member functions
template<typename T>
void OnData(const std::string& channelName, bool (T::* memberFunction)(FairMQMessagePtr& msg, int index))
{
fDataCallbacks = true;
fMsgInputs.insert(std::make_pair(channelName, [this, memberFunction](FairMQMessagePtr& msg, int index)
{
return (static_cast<T*>(this)->*memberFunction)(msg, index);
}));
if (find(fInputChannelKeys.begin(), fInputChannelKeys.end(), channelName) == fInputChannelKeys.end())
{
fInputChannelKeys.push_back(channelName);
}
}
void OnData(const std::string& channelName, InputMsgCallback callback)
{
fDataCallbacks = true;
fMsgInputs.insert(make_pair(channelName, callback));
if (find(fInputChannelKeys.begin(), fInputChannelKeys.end(), channelName) == fInputChannelKeys.end())
{
fInputChannelKeys.push_back(channelName);
}
}
// overload to easily bind member functions
template<typename T>
void OnData(const std::string& channelName, bool (T::* memberFunction)(FairMQParts& parts, int index))
{
fDataCallbacks = true;
fMultipartInputs.insert(std::make_pair(channelName, [this, memberFunction](FairMQParts& parts, int index)
{
return (static_cast<T*>(this)->*memberFunction)(parts, index);
}));
if (find(fInputChannelKeys.begin(), fInputChannelKeys.end(), channelName) == fInputChannelKeys.end())
{
fInputChannelKeys.push_back(channelName);
}
}
void OnData(const std::string& channelName, InputMultipartCallback callback)
{
fDataCallbacks = true;
fMultipartInputs.insert(make_pair(channelName, callback));
if (find(fInputChannelKeys.begin(), fInputChannelKeys.end(), channelName) == fInputChannelKeys.end())
{
fInputChannelKeys.push_back(channelName);
}
}
FairMQChannel& GetChannel(const std::string& channelName, const int index = 0)
try {
return fChannels.at(channelName).at(index);
} catch (const std::out_of_range& oor) {
LOG(error) << "out of range: " << oor.what();
LOG(error) << "requested channel has not been configured? check channel names/configuration.";
fRateLogging = false;
throw;
}
virtual void RegisterChannelEndpoints() {}
bool RegisterChannelEndpoint(const std::string& channelName, uint16_t minNumSubChannels = 1, uint16_t maxNumSubChannels = 1)
{
bool ok = fChannelRegistry.insert(std::make_pair(channelName, std::make_pair(minNumSubChannels, maxNumSubChannels))).second;
if (!ok)
{
LOG(warn) << "Registering channel: name already registered: \"" << channelName << "\"";
}
return ok;
}
void PrintRegisteredChannels()
{
if (fChannelRegistry.size() < 1)
{
std::cout << "no channels registered." << std::endl;
}
else
{
for (const auto& c : fChannelRegistry)
{
std::cout << c.first << ":" << c.second.first << ":" << c.second.second << std::endl;
}
}
}
void SetId(const std::string& id) { fId = id; }
std::string GetId() { return fId; }
const fair::mq::tools::Version GetVersion() const { return fVersion; }
void SetNumIoThreads(int numIoThreads) { fConfig->SetValue<int>("io-threads", numIoThreads);}
int GetNumIoThreads() const { return fConfig->GetValue<int>("io-threads"); }
void SetNetworkInterface(const std::string& networkInterface) { fConfig->SetValue<std::string>("network-interface", networkInterface); }
std::string GetNetworkInterface() const { return fConfig->GetValue<std::string>("network-interface"); }
void SetDefaultTransport(const std::string& name) { fConfig->SetValue<std::string>("transport", name); }
std::string GetDefaultTransport() const { return fConfig->GetValue<std::string>("transport"); }
void SetInitializationTimeoutInS(int initializationTimeoutInS) { fConfig->SetValue<int>("initialization-timeout", initializationTimeoutInS); }
int GetInitializationTimeoutInS() const { return fConfig->GetValue<int>("initialization-timeout"); }
void SetTransport(const std::string& transport) { fConfig->SetValue<std::string>("transport", transport); }
std::string GetTransportName() const { return fConfig->GetValue<std::string>("transport"); }
void SetRawCmdLineArgs(const std::vector<std::string>& args) { fRawCmdLineArgs = args; }
std::vector<std::string> GetRawCmdLineArgs() const { return fRawCmdLineArgs; }
void RunStateMachine()
{
CallStateChangeCallbacks(FairMQStateMachine::IDLE);
ProcessWork();
};
template<class Rep, class Period>
bool WaitFor(std::chrono::duration<Rep, Period> const& duration)
{
std::unique_lock<std::mutex> lock(fInterruptedMtx);
return !fInterruptedCV.wait_for(lock, duration, [&] { return fInterrupted.load(); }); // return true if no interruption happened
}
protected:
std::shared_ptr<FairMQTransportFactory> fTransportFactory;
std::unordered_map<fair::mq::Transport, std::shared_ptr<FairMQTransportFactory>> fTransports;
public:
std::unordered_map<std::string, std::vector<FairMQChannel>> fChannels;
std::unique_ptr<FairMQProgOptions> fInternalConfig;
FairMQProgOptions* fConfig;
void AddChannel(const std::string& channelName, const FairMQChannel& channel)
{
fConfig->AddChannel(channelName, channel);
}
protected:
std::string fId;
virtual void Init();
virtual void InitTask();
virtual void Run();
virtual void PreRun();
virtual bool ConditionalRun();
virtual void PostRun();
virtual void Pause();
virtual void ResetTask();
virtual void Reset();
private:
fair::mq::Transport fDefaultTransportType;
void InitWrapper();
void InitTaskWrapper();
void RunWrapper();
void PauseWrapper();
void ResetTaskWrapper();
void ResetWrapper();
void Unblock();
void Exit();
void AttachChannels(std::vector<FairMQChannel*>& chans);
bool AttachChannel(FairMQChannel& ch);
void HandleSingleChannelInput();
void HandleMultipleChannelInput();
void HandleMultipleTransportInput();
void PollForTransport(const FairMQTransportFactory* factory, const std::vector<std::string>& channelKeys);
bool HandleMsgInput(const std::string& chName, const InputMsgCallback& callback, int i);
bool HandleMultipartInput(const std::string& chName, const InputMultipartCallback& callback, int i);
void CreateOwnConfig();
bool fDataCallbacks;
std::unordered_map<std::string, InputMsgCallback> fMsgInputs;
std::unordered_map<std::string, InputMultipartCallback> fMultipartInputs;
std::unordered_map<fair::mq::Transport, std::vector<std::string>> fMultitransportInputs;
std::unordered_map<std::string, std::pair<uint16_t, uint16_t>> fChannelRegistry;
std::vector<std::string> fInputChannelKeys;
std::mutex fMultitransportMutex;
std::atomic<bool> fMultitransportProceed;
const fair::mq::tools::Version fVersion;
float fRate;
std::vector<std::string> fRawCmdLineArgs;
std::atomic<bool> fInterrupted;
std::condition_variable fInterruptedCV;
std::mutex fInterruptedMtx;
mutable std::atomic<bool> fRateLogging;
};
#endif /* FAIRMQDEVICE_H_ */
FairMQDevice::Pause
virtual void Pause()
Handles the PAUSE state.
Definition: FairMQDevice.cxx:627
FairMQDevice::GetTransportName
std::string GetTransportName() const
Gets the default transport name.
Definition: FairMQDevice.h:385
FairMQDevice::fTransports
std::unordered_map< fair::mq::Transport, std::shared_ptr< FairMQTransportFactory > > fTransports
Container for transports.
Definition: FairMQDevice.h:408
FairMQDevice::Run
virtual void Run()
Runs the device (to be overloaded in child classes)
Definition: FairMQDevice.cxx:603
FairMQChannel::Send
int Send(FairMQMessagePtr &msg, int sndTimeoutInMs=-1)
Definition: FairMQChannel.h:234
FairMQDevice::ConditionalRun
virtual bool ConditionalRun()
Called during RUNNING state repeatedly until it returns false or device state changes.
Definition: FairMQDevice.cxx:611
FairMQChannel::Receive
int Receive(FairMQMessagePtr &msg, int rcvTimeoutInMs=-1)
Definition: FairMQChannel.h:244
FairMQDevice::fConfig
FairMQProgOptions * fConfig
Pointer to config (internal or external)
Definition: FairMQDevice.h:413
FairMQTransportFactory
Definition: FairMQTransportFactory.h:28
FairMQDevice::SetTransport
void SetTransport(const std::string &transport)
Definition: FairMQDevice.h:383
FairMQDevice::Send
int Send(FairMQMessagePtr &msg, const std::string &channel, const int index=0, int sndTimeoutInMs=-1)
Definition: FairMQDevice.h:102
FairMQDevice::WaitFor
bool WaitFor(std::chrono::duration< Rep, Period > const &duration)
Definition: FairMQDevice.h:400
FairMQDevice::fInternalConfig
std::unique_ptr< FairMQProgOptions > fInternalConfig
Internal program options configuration.
Definition: FairMQDevice.h:412
FairMQChannel
Definition: FairMQChannel.h:27
FairMQDevice::Send
int64_t Send(FairMQParts &parts, const std::string &channel, const int index=0, int sndTimeoutInMs=-1)
Definition: FairMQDevice.h:133
FairMQProgOptions
Definition: FairMQProgOptions.h:37
FairMQDevice::PreRun
virtual void PreRun()
Called in the RUNNING state once before executing the Run()/ConditionalRun() method.
Definition: FairMQDevice.cxx:607
FairMQDevice::ResetTask
virtual void ResetTask()
Resets the user task (to be overloaded in child classes)
Definition: FairMQDevice.cxx:795
FairMQDevice::fId
std::string fId
Device ID.
Definition: FairMQDevice.h:421
FairMQDevice::CatchSignals
void CatchSignals()
Catches interrupt signals (SIGINT, SIGTERM)
FairMQDevice::AddTransport
std::shared_ptr< FairMQTransportFactory > AddTransport(const fair::mq::Transport transport)
Definition: FairMQDevice.cxx:637
FairMQDevice::~FairMQDevice
virtual ~FairMQDevice()
Default destructor.
Definition: FairMQDevice.cxx:832
FairMQDevice::Receive
int Receive(FairMQMessagePtr &msg, const std::string &channel, const int index=0, int rcvTimeoutInMs=-1)
Definition: FairMQDevice.h:113
FairMQDevice::SortChannel
void SortChannel(const std::string &name, const bool reindex=true)
Definition: FairMQDevice.cxx:314
FairMQDevice::WaitForInitialValidation
void WaitForInitialValidation() __attribute__((deprecated("This method will have no effect in future versions and will be removed. Instead subscribe for state changes and inspect configuration values.")))
Waits for the first initialization run to finish.
Definition: FairMQDevice.h:254
FairMQDevice::PostRun
virtual void PostRun()
Called in the RUNNING state once after executing the Run()/ConditionalRun() method.
Definition: FairMQDevice.cxx:616
FairMQDevice::fChannels
std::unordered_map< std::string, std::vector< FairMQChannel > > fChannels
Device channels.
Definition: FairMQDevice.h:411
FairMQStateMachine
Definition: FairMQStateMachine.h:27
FairMQDevice::Receive
int64_t Receive(FairMQParts &parts, const std::string &channel, const int index=0, int rcvTimeoutInMs=-1)
Definition: FairMQDevice.h:144
FairMQDevice::GetConfig
FairMQProgOptions * GetConfig() const
Get pointer to the config.
Definition: FairMQDevice.h:263
FairMQParts
FairMQParts is a lightweight convenience wrapper around a vector of unique pointers to FairMQMessage,...
Definition: FairMQParts.h:20
FairMQDevice::fTransportFactory
std::shared_ptr< FairMQTransportFactory > fTransportFactory
Default transport factory.
Definition: FairMQDevice.h:407
FairMQDevice::SetConfig
void SetConfig(FairMQProgOptions &config)
Assigns config to the device.
Definition: FairMQDevice.cxx:659
FairMQDevice::Reset
virtual void Reset()
Resets the device (can be overloaded in child classes)
Definition: FairMQDevice.cxx:824
FairMQDevice::SortSocketsByAddress
static bool SortSocketsByAddress(const FairMQChannel &lhs, const FairMQChannel &rhs)
Definition: FairMQDevice.cxx:309
FairMQDevice
Definition: FairMQDevice.h:46
FairMQDevice::Init
virtual void Init()
Additional user initialization (can be overloaded in child classes). Prefer to use InitTask().
Definition: FairMQDevice.cxx:198
FairMQMessage
Definition: FairMQMessage.h:20
FairMQDevice::InitTask
virtual void InitTask()
Task initialization (can be overloaded in child classes)
Definition: FairMQDevice.cxx:305
FairMQDevice::operator=
FairMQDevice operator=(const FairMQDevice &)=delete
Assignment operator (disabled)
FairMQDevice::Transport
auto Transport() const -> FairMQTransportFactory *
Getter for default transport factory.
Definition: FairMQDevice.h:159
FairMQDevice::LogSocketRates
virtual void LogSocketRates()
Outputs the socket transfer rates.
Definition: FairMQDevice.cxx:665
FairMQDevice::FairMQDevice
FairMQDevice()
Default constructor.
Definition: FairMQDevice.cxx:30
fair::mq::tools::Version
Definition: Version.h:22

privacy