Source · fairmq/
Device.hpublic header
Defines class fair::mq::Device, struct fair::mq::OngoingTransition.
Described in Channels and sockets, Devices, Device states, Threads in a device, Transports, 1-1: sampler and sink, 1-n-1: sampler, processors and sink, copypush: one message to several receivers, dds: run with DDS, multipart: several parts in one message, multiple-channels: data and broadcast, multiple-transports: zeromq and shmem, n-m: n senders to m receivers, qc: quality control on demand, readout: a readout chain, region: unmanaged regions, req-rep: request and reply, Run a benchmark, Configure channels, Control a device from code, Design a topology, Stop a topology without losing messages, Handle errors and timeouts, Move from the FairMQ* names, Poll several channels, Receive data with callbacks or loops, Send your own data types, Use several transports, States and transitions, Transfer codes and errors, Your first device, Overview.
1/********************************************************************************2 * Copyright (C) 2021-2023 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *3 * *4 * This software is distributed under the terms of the *5 * GNU Lesser General Public Licence (LGPL) version 3, *6 * copied verbatim in the file "LICENSE" *7 ********************************************************************************/8 9#ifndef FAIR_MQ_DEVICE_H10#define FAIR_MQ_DEVICE_H11 12// FairMQ13#include <fairmq/Channel.h>14#include <fairmq/Error.h>15#include <fairmq/Message.h>16#include <fairmq/Parts.h>17#include <fairmq/ProgOptions.h>18#include <fairmq/StateMachine.h>19#include <fairmq/StateQueue.h>20#include <fairmq/Tools.h>21#include <fairmq/TransportFactory.h>22#include <fairmq/TransportEnum.h>23#include <fairmq/UnmanagedRegion.h>24 25// logger26#include <fairlogger/Logger.h>27 28// std29#include <algorithm> // find30#include <atomic>31#include <chrono>32#include <cstddef>33#include <functional>34#include <memory> // unique_ptr35#include <mutex>36#include <stdexcept>37#include <string>38#include <unordered_map>39#include <utility> // pair40#include <vector>41 42namespace fair::mq {43 44using ChannelMap = std::unordered_map<std::string, std::vector<Channel>>;45 46struct OngoingTransition : std::runtime_error47{48 using std::runtime_error::runtime_error;49};50 51using InputMsgCallback = std::function<bool(MessagePtr&, int)>;52 53using InputMultipartCallback = std::function<bool(Parts&, int)>;54 55class Device56{57 friend class Channel;58 59 public:60 Device();61 Device(ProgOptions& config);62 Device(tools::Version version);63 Device(ProgOptions& config, tools::Version version);64 65 private:66 Device(ProgOptions* config, tools::Version version);67 68 public:69 Device(const Device&) = delete;70 Device(Device&&) = delete;71 Device& operator=(const Device&) = delete;72 Device& operator=(Device&&) = delete;73 virtual ~Device();74 75 /// Outputs the socket transfer rates76 virtual void LogSocketRates();77 78 template<typename Serializer, typename DataType, typename... Args>79 [[deprecated]] void Serialize(Message& msg, DataType&& data, Args&&... args) const80 {81 Serializer().Serialize(msg, std::forward<DataType>(data), std::forward<Args>(args)...);82 }83 84 template<typename Deserializer, typename DataType, typename... Args>85 [[deprecated]] void Deserialize(Message& msg, DataType&& data, Args&&... args) const86 {87 Deserializer().Deserialize(msg, std::forward<DataType>(data), std::forward<Args>(args)...);88 }89 90 /// Send `m` on `chan` at index `i`91 /// @param m reference to MessagePtr/Parts/vector<MessagePtr>92 /// @param chan channel name93 /// @param i channel index94 /// @return Number of queued bytes,95 /// TransferCode::timeout if timed out,96 /// TransferCode::error if there was an error,97 /// TransferCode::interrupted if interrupted (e.g. by requested state change)98 template<typename M>99 std::enable_if_t<is_transferrable<M>::value, int64_t>100 Send(M& m, const std::string& channel, const int index = 0)101 {102 return GetChannel(channel, index).Send(m);103 }104 105 /// Receive `m` on `chan` at index `i`106 /// @param m reference to MessagePtr/Parts/vector<MessagePtr>107 /// @param chan channel name108 /// @param i channel index109 /// @return Number of received bytes,110 /// TransferCode::timeout if timed out,111 /// TransferCode::error if there was an error,112 /// TransferCode::interrupted if interrupted (e.g. by requested state change)113 template<typename M>114 std::enable_if_t<is_transferrable<M>::value, int64_t>115 Receive(M& m, const std::string& channel, const int index = 0)116 {117 return GetChannel(channel, index).Receive(m);118 }119 120 /// Send `m` on `chan` at index `i`121 /// @param m reference to MessagePtr/Parts/vector<MessagePtr>122 /// @param chan channel name123 /// @param i channel index124 /// @param sndTimeoutMs send timeout in ms,125 /// -1 will wait forever (or until interrupt (e.g. via state change)),126 /// 0 will not wait (return immediately if cannot send)127 /// @return Number of queued bytes,128 /// TransferCode::timeout if timed out,129 /// TransferCode::error if there was an error,130 /// TransferCode::interrupted if interrupted (e.g. by requested state change)131 template<typename M>132 std::enable_if_t<is_transferrable<M>::value, int64_t>133 Send(M& m, const std::string& channel, const int index, int sndTimeoutMs)134 {135 return GetChannel(channel, index).Send(m, sndTimeoutMs);136 }137 138 /// Receive `m` on `chan` at index `i`139 /// @param m reference to MessagePtr/Parts/vector<MessagePtr>140 /// @param chan channel name141 /// @param i channel index142 /// @param rcvTimeoutMs receive timeout in ms,143 /// -1 will wait forever (or until interrupt (e.g. via state change),144 /// 0 will not wait (return immediately if cannot receive)145 /// @return Number of received bytes,146 /// TransferCode::timeout if timed out,147 /// TransferCode::error if there was an error,148 /// TransferCode::interrupted if interrupted (e.g. by requested state change)149 template<typename M>150 std::enable_if_t<is_transferrable<M>::value, int64_t>151 Receive(M& m, const std::string& channel, const int index, int rcvTimeoutMs)152 {153 return GetChannel(channel, index).Receive(m, rcvTimeoutMs);154 }155 156 /// @brief Getter for default transport factory157 auto Transport() const -> TransportFactory* { return fTransportFactory.get(); }158 159 // creates message with the default device transport160 template<typename... Args>161 MessagePtr NewMessage(Args&&... args)162 {163 return Transport()->CreateMessage(std::forward<Args>(args)...);164 }165 166 // creates message with the transport of the specified channel167 template<typename... Args>168 MessagePtr NewMessageFor(const std::string& channel, int index, Args&&... args)169 {170 return GetChannel(channel, index).NewMessage(std::forward<Args>(args)...);171 }172 173 // creates a message that will not be cleaned up after transfer, with the default device174 // transport175 template<typename T>176 MessagePtr NewStaticMessage(const T& data)177 {178 return Transport()->NewStaticMessage(data);179 }180 181 // creates a message that will not be cleaned up after transfer, with the transport of the182 // specified channel183 template<typename T>184 MessagePtr NewStaticMessageFor(const std::string& channel, int index, const T& data)185 {186 return GetChannel(channel, index).NewStaticMessage(data);187 }188 189 // creates a message with a copy of the provided data, with the default device transport190 template<typename T>191 MessagePtr NewSimpleMessage(const T& data)192 {193 return Transport()->NewSimpleMessage(data);194 }195 196 // creates a message with a copy of the provided data, with the transport of the specified197 // channel198 template<typename T>199 MessagePtr NewSimpleMessageFor(const std::string& channel, int index, const T& data)200 {201 return GetChannel(channel, index).NewSimpleMessage(data);202 }203 204 // creates unamanaged region with the default device transport205 template<typename... Args>206 UnmanagedRegionPtr NewUnmanagedRegion(Args&&... args)207 {208 return Transport()->CreateUnmanagedRegion(std::forward<Args>(args)...);209 }210 211 // creates unmanaged region with the transport of the specified channel212 template<typename... Args>213 UnmanagedRegionPtr NewUnmanagedRegionFor(const std::string& channel, int index, Args&&... args)214 {215 return GetChannel(channel, index).NewUnmanagedRegion(std::forward<Args>(args)...);216 }217 218 template<typename... Ts>219 PollerPtr NewPoller(const Ts&... inputs)220 {221 std::vector<std::string> chans{inputs...};222 223 // if more than one channel provided, check compatibility224 if (chans.size() > 1) {225 mq::Transport type = GetChannel(chans.at(0), 0).Transport()->GetType();226 227 for (unsigned int i = 1; i < chans.size(); ++i) {228 if (type != GetChannel(chans.at(i), 0).Transport()->GetType()) {229 LOG(error) << "poller failed: different transports within same poller are not "230 "yet supported. Going to ERROR state.";231 throw std::runtime_error("poller failed: different transports within same "232 "poller are not yet supported.");233 }234 }235 }236 237 return GetChannel(chans.at(0), 0).Transport()->CreatePoller(GetChannels(), chans);238 }239 240 PollerPtr NewPoller(const std::vector<Channel*>& channels)241 {242 // if more than one channel provided, check compatibility243 if (channels.size() > 1) {244 mq::Transport type = channels.at(0)->Transport()->GetType();245 246 for (unsigned int i = 1; i < channels.size(); ++i) {247 if (type != channels.at(i)->Transport()->GetType()) {248 LOG(error) << "poller failed: different transports within same poller are not "249 "yet supported. Going to ERROR state.";250 throw std::runtime_error("poller failed: different transports within same "251 "poller are not yet supported.");252 }253 }254 }255 256 return channels.at(0)->Transport()->CreatePoller(channels);257 }258 259 /// Adds a transport to the device if it doesn't exist260 /// @param transport Transport string ("zeromq"/"shmem")261 std::shared_ptr<TransportFactory> AddTransport(mq::Transport transport);262 263 /// Assigns config to the device264 void SetConfig(ProgOptions& config);265 /// Get pointer to the config266 ProgOptions* GetConfig() const { return fConfig; }267 268 // overload to easily bind member functions269 template<typename T>270 void OnData(const std::string& channelName,271 bool (T::*memberFunction)(MessagePtr& msg, int index))272 {273 fDataCallbacks = true;274 fMsgInputs.insert(275 std::make_pair(channelName, [this, memberFunction](MessagePtr& msg, int index) {276 return (static_cast<T*>(this)->*memberFunction)(msg, index);277 }));278 279 if (find(fInputChannelKeys.begin(), fInputChannelKeys.end(), channelName)280 == fInputChannelKeys.end()) {281 fInputChannelKeys.push_back(channelName);282 }283 }284 285 void OnData(const std::string& channelName, InputMsgCallback callback)286 {287 fDataCallbacks = true;288 fMsgInputs.insert(make_pair(channelName, callback));289 290 if (find(fInputChannelKeys.begin(), fInputChannelKeys.end(), channelName)291 == fInputChannelKeys.end()) {292 fInputChannelKeys.push_back(channelName);293 }294 }295 296 // overload to easily bind member functions297 template<typename T>298 void OnData(const std::string& channelName, bool (T::*memberFunction)(Parts& parts, int index))299 {300 fDataCallbacks = true;301 fMultipartInputs.insert(302 std::make_pair(channelName, [this, memberFunction](Parts& parts, int index) {303 return (static_cast<T*>(this)->*memberFunction)(parts, index);304 }));305 306 if (find(fInputChannelKeys.begin(), fInputChannelKeys.end(), channelName)307 == fInputChannelKeys.end()) {308 fInputChannelKeys.push_back(channelName);309 }310 }311 312 void OnData(const std::string& channelName, InputMultipartCallback callback)313 {314 fDataCallbacks = true;315 fMultipartInputs.insert(make_pair(channelName, callback));316 317 if (find(fInputChannelKeys.begin(), fInputChannelKeys.end(), channelName)318 == fInputChannelKeys.end()) {319 fInputChannelKeys.push_back(channelName);320 }321 }322 323 Channel& GetChannel(const std::string& channelName, const int index = 0)324 try {325 return GetChannels().at(channelName).at(index);326 } catch (const std::out_of_range& oor) {327 LOG(error) << "GetChannel(): '" << channelName << "[" << index << "]' does not exist.";328 throw;329 }330 331 size_t GetNumSubChannels(const std::string& channelName)332 try {333 return GetChannels().at(channelName).size();334 } catch (const std::out_of_range& oor) {335 LOG(error) << "GetNumSubChannels(): '" << channelName << "' does not exist.";336 throw;337 }338 339 /// @brief Get numbers of connected peers for the given channel340 /// @param name channel name341 /// @param index sub-channel342 unsigned long GetNumberOfConnectedPeers(const std::string& channelName, int index = 0)343 {344 return GetChannel(channelName, index).GetNumberOfConnectedPeers();345 }346 347 virtual void RegisterChannelEndpoints() {}348 349 bool RegisterChannelEndpoint(const std::string& channelName,350 uint16_t minNumSubChannels = 1,351 uint16_t maxNumSubChannels = 1)352 {353 bool ok = fChannelRegistry354 .insert(std::make_pair(channelName,355 std::make_pair(minNumSubChannels, maxNumSubChannels)))356 .second;357 if (!ok) {358 LOG(warn) << "Registering channel: name already registered: \"" << channelName << "\"";359 }360 return ok;361 }362 363 void PrintRegisteredChannels()364 {365 if (fChannelRegistry.empty()) {366 LOGV(info, verylow) << "no channels registered.";367 } else {368 for (const auto& c : fChannelRegistry) {369 LOGV(info, verylow) << c.first << ":" << c.second.first << ":" << c.second.second;370 }371 }372 }373 374 void SetId(const std::string& id) { fId = id; }375 std::string GetId() { return fId; }376 377 const tools::Version GetVersion() const { return fVersion; }378 379 void SetNumIoThreads(int numIoThreads) { fConfig->SetProperty("io-threads", numIoThreads); }380 int GetNumIoThreads() const381 {382 return fConfig->GetProperty<int>("io-threads", DefaultIOThreads);383 }384 385 void SetNetworkInterface(const std::string& networkInterface)386 {387 fConfig->SetProperty("network-interface", networkInterface);388 }389 std::string GetNetworkInterface() const390 {391 return fConfig->GetProperty<std::string>("network-interface", DefaultNetworkInterface);392 }393 394 void SetDefaultTransport(const std::string& name) { fConfig->SetProperty("transport", name); }395 std::string GetDefaultTransport() const396 {397 return fConfig->GetProperty<std::string>("transport", DefaultTransportName);398 }399 400 void SetInitTimeoutInS(int initTimeoutInS)401 {402 fConfig->SetProperty("init-timeout", initTimeoutInS);403 }404 int GetInitTimeoutInS() const405 {406 return fConfig->GetProperty<int>("init-timeout", DefaultInitTimeout);407 }408 409 /// Sets the default transport for the device410 /// @param transport Transport string ("zeromq"/"shmem")411 void SetTransport(const std::string& transport)412 {413 fConfig->SetProperty("transport", transport);414 }415 /// Gets the default transport name416 std::string GetTransportName() const417 {418 return fConfig->GetProperty<std::string>("transport", DefaultTransportName);419 }420 421 void SetRawCmdLineArgs(const std::vector<std::string>& args) { fRawCmdLineArgs = args; }422 std::vector<std::string> GetRawCmdLineArgs() const { return fRawCmdLineArgs; }423 424 void RunStateMachine() { fStateMachine.ProcessWork(); };425 426 /// Wait for the supplied amount of time or for interruption.427 /// If interrupted, returns false, otherwise true.428 /// @param duration wait duration429 template<typename Rep, typename Period>430 bool WaitFor(std::chrono::duration<Rep, Period> const& duration)431 {432 return !fStateMachine.WaitForPendingStateFor(433 std::chrono::duration_cast<std::chrono::milliseconds>(duration).count());434 }435 436 protected:437 std::shared_ptr<TransportFactory> fTransportFactory; ///< Default transport factory438 std::unordered_map<mq::Transport, std::shared_ptr<TransportFactory>>439 fTransports; ///< Container for transports440 441 public:442 [[deprecated("Use GetChannels() instead.")]]443 std::unordered_map<std::string, std::vector<Channel>> fChannels; ///< Device channels444 std::unordered_map<std::string, std::vector<Channel>>& GetChannels()445 {446#pragma GCC diagnostic push447#pragma GCC diagnostic ignored "-Wdeprecated-declarations"448 return fChannels;449#pragma GCC diagnostic pop450 }451 std::unordered_map<std::string, std::vector<Channel>> const& GetChannels() const452 {453#pragma GCC diagnostic push454#pragma GCC diagnostic ignored "-Wdeprecated-declarations"455 return fChannels;456#pragma GCC diagnostic pop457 }458 459 std::unique_ptr<ProgOptions> fInternalConfig; ///< Internal program options configuration460 ProgOptions* fConfig; ///< Pointer to config (internal or external)461 462 void AddChannel(const std::string& name, Channel&& channel)463 {464 fConfig->AddChannel(name, channel);465 }466 467 protected:468 std::string fId; ///< Device ID469 470 /// Additional user initialization (can be overloaded in child classes). Prefer to use471 /// InitTask().472 virtual void Init() {}473 474 virtual void Bind() {}475 476 virtual void Connect() {}477 478 /// Task initialization (can be overloaded in child classes)479 virtual void InitTask() {}480 481 /// Runs the device (to be overloaded in child classes)482 virtual void Run() {}483 484 /// Called in the RUNNING state once before executing the Run()/ConditionalRun() method485 virtual void PreRun() {}486 487 /// Called during RUNNING state repeatedly until it returns false or device state changes488 virtual bool ConditionalRun() { return false; }489 490 /// Called in the RUNNING state once after executing the Run()/ConditionalRun() method491 virtual void PostRun() {}492 493 /// Resets the user task (to be overloaded in child classes)494 virtual void ResetTask() {}495 496 /// Resets the device (can be overloaded in child classes)497 virtual void Reset() {}498 499 public:500 /// @brief Request a device state transition501 /// @param transition state transition502 /// @return whether the transition was successfully scheduled503 ///504 /// The state transition may not happen immediately, but when the current state evaluates the505 /// pending transition event and terminates. In other words, the device states are scheduled506 /// cooperatively.507 [[nodiscard]] bool ChangeState(const Transition transition)508 {509 return fStateMachine.ChangeState(transition);510 }511 /// @brief Request a device state transition512 /// @param transition state transition513 /// @return whether the transition was successfully scheduled514 ///515 /// The state transition may not happen immediately, but when the current state evaluates the516 /// pending transition event and terminates. In other words, the device states are scheduled517 /// cooperatively.518 [[nodiscard]] bool ChangeState(const std::string& transition)519 {520 return fStateMachine.ChangeState(GetTransition(transition));521 }522 /// @brief Request a device state transition523 /// @param transition state transition524 /// @throws when the transition could not have been scheduled525 ///526 /// Throwing version of Device::ChangeState().527 void ChangeStateOrThrow(Transition transition)528 {529 if(!ChangeState(transition)) {530 auto const err = MakeErrorCode(ErrorCode::DeviceChangeStateFailed);531 throw std::system_error(err.value(),532 err.category(),533 tools::ToString("Invalid transition: ", transition));534 }535 }536 /// @brief Request a device state transition537 /// @param transition state transition538 /// @throws when the transition could not have been scheduled539 ///540 /// Throwing version of Device::ChangeState().541 void ChangeStateOrThrow(std::string const& transition)542 {543 ChangeStateOrThrow(GetTransition(transition));544 }545 546 /// @brief waits for the next state (any) to occur547 State WaitForNextState() { return fStateQueue.WaitForNext(); }548 /// @brief waits for the specified state to occur549 /// @param state state to wait for550 void WaitForState(State state) { fStateQueue.WaitForState(state); }551 /// @brief waits for the specified state to occur552 /// @param state state to wait for553 void WaitForState(const std::string& state) { WaitForState(GetState(state)); }554 555 /// @brief Subscribe with a callback to state changes556 /// @param key id to identify your subscription557 /// @param callback callback (called with the new state as the parameter)558 ///559 /// The callback is called at the beginning of a new state.560 /// The callback is called from the thread the state is running in.561 void SubscribeToStateChange(const std::string& key, std::function<void(const State)> callback)562 {563 fStateMachine.SubscribeToStateChange(key, callback);564 }565 /// @brief Unsubscribe from state changes566 /// @param key id (that was used when subscribing)567 void UnsubscribeFromStateChange(const std::string& key)568 {569 fStateMachine.UnsubscribeFromStateChange(key);570 }571 572 /// @brief Subscribe with a callback to incoming state transitions573 /// @param key id to identify your subscription574 /// @param callback callback (called with the incoming transition as the parameter)575 /// The callback is called when new transition is initiated.576 /// The callback is called from the thread that initiates the transition (via ChangeState).577 void SubscribeToNewTransition(const std::string& key,578 std::function<void(const Transition)> callback)579 {580 fStateMachine.SubscribeToNewTransition(key, callback);581 }582 /// @brief Unsubscribe from state transitions583 /// @param key id (that was used when subscribing)584 void UnsubscribeFromNewTransition(const std::string& key)585 {586 fStateMachine.UnsubscribeFromNewTransition(key);587 }588 589 /// @brief Returns true if a new state has been requested, signaling the current handler to590 /// stop.591 bool NewStatePending() const { return fStateMachine.NewStatePending(); }592 593 /// @brief Returns the current state594 State GetCurrentState() const { return fStateMachine.GetCurrentState(); }595 /// @brief Returns the name of the current state as a string596 std::string GetCurrentStateName() const { return fStateMachine.GetCurrentStateName(); }597 598 /// @brief Returns name of the given state as a string599 /// @param state state600 [[deprecated("Use fair::mq::GetStateName from <fairmq/States.h> directly")]]601 static std::string GetStateName(State state) { return fair::mq::GetStateName(state); }602 /// @brief Returns name of the given transition as a string603 /// @param transition transition604 [[deprecated("Use fair::mq::GetTransitionName from <fairmq/States.h> directly")]]605 static std::string GetTransitionName(Transition transition)606 {607 return fair::mq::GetTransitionName(transition);608 }609 610 static constexpr const char* DefaultId = "";611 static constexpr int DefaultIOThreads = 1;612 static constexpr const char* DefaultTransportName = "zeromq";613 static constexpr mq::Transport DefaultTransportType = mq::Transport::ZMQ;614 static constexpr const char* DefaultNetworkInterface = "default";615 static constexpr int DefaultInitTimeout = 120;616 static constexpr float DefaultRate = 0.;617 static constexpr const char* DefaultSession = "default";618 619 private:620 mq::Transport fDefaultTransportType; ///< Default transport for the device621 StateMachine fStateMachine;622 623 /// Handles the initialization624 void InitWrapper();625 /// Initializes binding channels626 void BindWrapper();627 /// Initializes connecting channels628 void ConnectWrapper();629 /// Handles the InitTask() method630 void InitTaskWrapper();631 /// Handles the Run() method632 void RunWrapper();633 /// Handles the ResetTask() method634 void ResetTaskWrapper();635 /// Handles the Reset() method636 void ResetWrapper();637 638 /// Notifies transports to cease any blocking activity639 void InterruptTransports();640 641 /// Notifies transports to resume any blocking activity642 void ResumeTransports();643 644 /// Shuts down the transports and the device645 void Exit() {}646 647 /// Attach (bind/connect) channels in the list648 void AttachChannels(std::vector<Channel*>& chans);649 bool AttachChannel(Channel& ch);650 651 void HandleSingleChannelInput();652 void HandleMultipleChannelInput();653 void HandleMultipleTransportInput();654 void PollForTransport(const TransportFactory* factory,655 const std::vector<std::string>& channelKeys);656 657 bool HandleMsgInput(const std::string& chName, const InputMsgCallback& callback, int i);658 bool HandleMultipartInput(const std::string& chName,659 const InputMultipartCallback& callback,660 int i);661 662 std::vector<Channel*> fUninitializedBindingChannels;663 std::vector<Channel*> fUninitializedConnectingChannels;664 665 bool fDataCallbacks;666 std::unordered_map<std::string, InputMsgCallback> fMsgInputs;667 std::unordered_map<std::string, InputMultipartCallback> fMultipartInputs;668 std::unordered_map<mq::Transport, std::vector<std::string>> fMultitransportInputs;669 std::unordered_map<std::string, std::pair<uint16_t, uint16_t>> fChannelRegistry;670 std::vector<std::string> fInputChannelKeys;671 std::mutex fMultitransportMutex;672 std::atomic<bool> fMultitransportProceed;673 674 const tools::Version fVersion;675 float fRate; ///< Rate limiting for ConditionalRun676 int fInitializationTimeoutInS;677 std::vector<std::string> fRawCmdLineArgs;678 679 StateQueue fStateQueue;680 681 std::mutex fTransportMtx; ///< guards access to transports container682};683 684} // namespace fair::mq685 686using FairMQChannelMap [[deprecated("Use fair::mq::ChannelMap")]] = fair::mq::ChannelMap;687using InputMsgCallback [[deprecated("Use fair::mq::InputMsgCallback")]] = fair::mq::InputMsgCallback;688using InputMultipartCallback [[deprecated("Use fair::mq::InputMultipartCallback")]] = fair::mq::InputMultipartCallback;689using FairMQDevice [[deprecated("Use fair::mq::Device")]] = fair::mq::Device;690 691#endif /* FAIR_MQ_DEVICE_H */