Device()No comment in the header yet.
API · fair::mq
Deviceclass
- Inherited by
fair::mq::BenchmarkSampler,fair::mq::Merger,fair::mq::Multiplier,fair::mq::Proxy,fair::mq::Sink,fair::mq::Splitter
No comment in the header yet.
Functions
Device(ProgOptions& config)No comment in the header yet.
Device(tools::Version version)No comment in the header yet.
Device(ProgOptions& config, tools::Version version)No comment in the header yet.
Device(const Device&) = deleteNo comment in the header yet.
Device(Device&&) = deleteNo comment in the header yet.
Device& operator=(const Device&) = deleteNo comment in the header yet.
Device& operator=(Device&&) = deleteNo comment in the header yet.
virtual ~Device()TODO: Remove this once Device::fChannels is no longer public.
virtual void LogSocketRates()Outputs the socket transfer rates.
template<typename Serializer, typename DataType, typename... Args>
void Serialize(Message& msg, DataType&& data, Args&&... args) constNo comment in the header yet.
template<typename Deserializer, typename DataType, typename... Args>
void Deserialize(Message& msg, DataType&& data, Args&&... args) constNo comment in the header yet.
template<typename M>
std::enable_if_t<is_transferrable<M>::value, int64_t> Send(
M& m,
const std::string& channel,
const int index = 0
)Send m on chan at index i
m- reference to MessagePtr/Parts/vector<MessagePtr>
chan- channel name
i- channel index
- Returns
- Number of queued bytes, TransferCode::timeout if timed out, TransferCode::error if there was an error, TransferCode::interrupted if interrupted (e.g. by requested state change)
template<typename M>
std::enable_if_t<is_transferrable<M>::value, int64_t> Receive(
M& m,
const std::string& channel,
const int index = 0
)Receive m on chan at index i
m- reference to MessagePtr/Parts/vector<MessagePtr>
chan- channel name
i- channel index
- Returns
- Number of received bytes, TransferCode::timeout if timed out, TransferCode::error if there was an error, TransferCode::interrupted if interrupted (e.g. by requested state change)
template<typename M>
std::enable_if_t<is_transferrable<M>::value, int64_t> Send(
M& m,
const std::string& channel,
const int index,
int sndTimeoutMs
)Send m on chan at index i
m- reference to MessagePtr/Parts/vector<MessagePtr>
chan- channel name
i- channel index
sndTimeoutMs- send timeout in ms, -1 will wait forever (or until interrupt (e.g. via state change)), 0 will not wait (return immediately if cannot send)
- Returns
- Number of queued bytes, TransferCode::timeout if timed out, TransferCode::error if there was an error, TransferCode::interrupted if interrupted (e.g. by requested state change)
template<typename M>
std::enable_if_t<is_transferrable<M>::value, int64_t> Receive(
M& m,
const std::string& channel,
const int index,
int rcvTimeoutMs
)Receive m on chan at index i
m- reference to MessagePtr/Parts/vector<MessagePtr>
chan- channel name
i- channel index
rcvTimeoutMs- receive timeout in ms, -1 will wait forever (or until interrupt (e.g. via state change), 0 will not wait (return immediately if cannot receive)
- Returns
- Number of received bytes, TransferCode::timeout if timed out, TransferCode::error if there was an error, TransferCode::interrupted if interrupted (e.g. by requested state change)
auto Transport() const -> TransportFactory *Getter for default transport factory.
template<typename... Args>
MessagePtr NewMessage(Args&&... args)No comment in the header yet.
template<typename... Args>
MessagePtr NewMessageFor(
const std::string& channel,
int index,
Args&&... args
)No comment in the header yet.
template<typename T>
MessagePtr NewStaticMessage(const T& data)No comment in the header yet.
template<typename T>
MessagePtr NewStaticMessageFor(
const std::string& channel,
int index,
const T& data
)No comment in the header yet.
template<typename T>
MessagePtr NewSimpleMessage(const T& data)No comment in the header yet.
template<typename T>
MessagePtr NewSimpleMessageFor(
const std::string& channel,
int index,
const T& data
)No comment in the header yet.
template<typename... Args>
UnmanagedRegionPtr NewUnmanagedRegion(Args&&... args)No comment in the header yet.
template<typename... Args>
UnmanagedRegionPtr NewUnmanagedRegionFor(
const std::string& channel,
int index,
Args&&... args
)No comment in the header yet.
template<typename... Ts>
PollerPtr NewPoller(const Ts&... inputs)No comment in the header yet.
PollerPtr NewPoller(const std::vector<Channel*>& channels)No comment in the header yet.
std::shared_ptr<TransportFactory> AddTransport(mq::Transport transport)Adds a transport to the device if it doesn't exist
transport- Transport string ("zeromq"/"shmem")
void SetConfig(ProgOptions& config)Assigns config to the device.
ProgOptions* GetConfig() constGet pointer to the config.
template<typename T>
void OnData(
const std::string& channelName,
bool(T::*)(MessagePtr&msg, int index) memberFunction
)No comment in the header yet.
void OnData(const std::string& channelName, InputMsgCallback callback)No comment in the header yet.
template<typename T>
void OnData(
const std::string& channelName,
bool(T::*)(Parts&parts, int index) memberFunction
)No comment in the header yet.
void OnData(const std::string& channelName, InputMultipartCallback callback)No comment in the header yet.
Channel& GetChannel(const std::string& channelName, const int index = 0)No comment in the header yet.
size_t GetNumSubChannels(const std::string& channelName)No comment in the header yet.
unsigned long GetNumberOfConnectedPeers(
const std::string& channelName,
int index = 0
)Get numbers of connected peers for the given channel.
name- channel name
index- sub-channel
virtual void RegisterChannelEndpoints()No comment in the header yet.
bool RegisterChannelEndpoint(
const std::string& channelName,
uint16_t minNumSubChannels = 1,
uint16_t maxNumSubChannels = 1
)No comment in the header yet.
void PrintRegisteredChannels()No comment in the header yet.
void SetId(const std::string& id)No comment in the header yet.
std::string GetId()No comment in the header yet.
const tools::Version GetVersion() constNo comment in the header yet.
void SetNumIoThreads(int numIoThreads)No comment in the header yet.
int GetNumIoThreads() constNo comment in the header yet.
void SetNetworkInterface(const std::string& networkInterface)No comment in the header yet.
std::string GetNetworkInterface() constNo comment in the header yet.
void SetDefaultTransport(const std::string& name)No comment in the header yet.
std::string GetDefaultTransport() constNo comment in the header yet.
void SetInitTimeoutInS(int initTimeoutInS)No comment in the header yet.
int GetInitTimeoutInS() constNo comment in the header yet.
void SetTransport(const std::string& transport)Sets the default transport for the device
transport- Transport string ("zeromq"/"shmem")
std::string GetTransportName() constGets the default transport name.
void SetRawCmdLineArgs(const std::vector<std::string>& args)No comment in the header yet.
std::vector<std::string> GetRawCmdLineArgs() constNo comment in the header yet.
void RunStateMachine()No comment in the header yet.
template<typename Rep, typename Period>
bool WaitFor(std::chrono::duration<Rep, Period> const& duration)Wait for the supplied amount of time or for interruption. If interrupted, returns false, otherwise true.
duration- wait duration
std::unordered_map<std::string, std::vector<Channel>>& GetChannels()No comment in the header yet.
std::unordered_map<std::string, std::vector<Channel>> const& GetChannels() constNo comment in the header yet.
void AddChannel(const std::string& name, Channel&& channel)No comment in the header yet.
bool ChangeState(const Transition transition)Request a device state transition.
The state transition may not happen immediately, but when the current state evaluates the pending transition event and terminates. In other words, the device states are scheduled cooperatively.
transition- state transition
- Returns
- whether the transition was successfully scheduled
bool ChangeState(const std::string& transition)Request a device state transition.
The state transition may not happen immediately, but when the current state evaluates the pending transition event and terminates. In other words, the device states are scheduled cooperatively.
transition- state transition
- Returns
- whether the transition was successfully scheduled
void ChangeStateOrThrow(Transition transition)Request a device state transition.
Throwing version of Device::ChangeState().
transition- state transition
whenthrows- the transition could not have been scheduled
void ChangeStateOrThrow(std::string const& transition)Request a device state transition.
Throwing version of Device::ChangeState().
transition- state transition
whenthrows- the transition could not have been scheduled
State WaitForNextState()waits for the next state (any) to occur
void WaitForState(State state)waits for the specified state to occur
state- state to wait for
void WaitForState(const std::string& state)waits for the specified state to occur
state- state to wait for
void SubscribeToStateChange(
const std::string& key,
std::function<void(const State)> callback
)Subscribe with a callback to state changes.
The callback is called at the beginning of a new state. The callback is called from the thread the state is running in.
key- id to identify your subscription
callback- callback (called with the new state as the parameter)
void UnsubscribeFromStateChange(const std::string& key)Unsubscribe from state changes.
key- id (that was used when subscribing)
void SubscribeToNewTransition(
const std::string& key,
std::function<void(const Transition)> callback
)Subscribe with a callback to incoming state transitions.
key- id to identify your subscription
callback- callback (called with the incoming transition as the parameter) The callback is called when new transition is initiated. The callback is called from the thread that initiates the transition (via ChangeState).
void UnsubscribeFromNewTransition(const std::string& key)Unsubscribe from state transitions.
key- id (that was used when subscribing)
bool NewStatePending() constReturns true if a new state has been requested, signaling the current handler to stop.
State GetCurrentState() constReturns the current state.
std::string GetCurrentStateName() constReturns the name of the current state as a string.
Static functions
static std::string GetStateName(State state)Returns name of the given state as a string.
state- state
static std::string GetTransitionName(Transition transition)Returns name of the given transition as a string.
transition- transition
Static data members
static constexpr const char* DefaultId = ""No comment in the header yet.
static constexpr int DefaultIOThreads = 1No comment in the header yet.
static constexpr const char* DefaultTransportName = "zeromq"No comment in the header yet.
static constexpr mq::Transport DefaultTransportType = mq::Transport::ZMQNo comment in the header yet.
static constexpr const char* DefaultNetworkInterface = "default"No comment in the header yet.
static constexpr int DefaultInitTimeout = 120No comment in the header yet.
static constexpr float DefaultRate = 0.No comment in the header yet.
static constexpr const char* DefaultSession = "default"No comment in the header yet.
Data members
std::unordered_map<std::string, std::vector<Channel>> fChannelsDevice channels.
std::unique_ptr<ProgOptions> fInternalConfigInternal program options configuration.
ProgOptions* fConfigPointer to config (internal or external)
Protected functions
virtual void Init()Additional user initialization (can be overloaded in child classes). Prefer to use InitTask().
virtual void Bind()No comment in the header yet.
virtual void Connect()No comment in the header yet.
virtual void InitTask()Task initialization (can be overloaded in child classes)
virtual void Run()Runs the device (to be overloaded in child classes)
virtual void PreRun()Called in the RUNNING state once before executing the Run()/ConditionalRun() method.
virtual bool ConditionalRun()Called during RUNNING state repeatedly until it returns false or device state changes.
virtual void PostRun()Called in the RUNNING state once after executing the Run()/ConditionalRun() method.
virtual void ResetTask()Resets the user task (to be overloaded in child classes)
virtual void Reset()Resets the device (can be overloaded in child classes)
Protected data members
std::shared_ptr<TransportFactory> fTransportFactoryDefault transport factory.
std::unordered_map<mq::Transport, std::shared_ptr<TransportFactory>> fTransportsContainer for transports.
std::string fIdDevice ID.