Source · fairmq/
Channel.hpublic header
Defines class fair::mq::Channel, struct fair::mq::Channel::ChannelConfigurationError.
Described in Channels and sockets, Messages and multipart, Threads in a device, multiple-transports: zeromq and shmem, n-m: n senders to m receivers, readout: a readout chain, req-rep: request and reply, Use channels without a device, Design a topology, Stop a topology without losing messages, Handle errors and timeouts, Use several transports, Channel properties, Overview.
1/********************************************************************************2 * Copyright (C) 2021-2022 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_CHANNEL_H10#define FAIR_MQ_CHANNEL_H11 12#include <fairmq/Message.h>13#include <fairmq/Parts.h>14#include <fairmq/Properties.h>15#include <fairmq/Socket.h>16#include <fairmq/TransportFactory.h>17#include <fairmq/TransportEnum.h>18#include <fairmq/UnmanagedRegion.h>19 20#include <cstdint> // int64_t21#include <memory> // unique_ptr, shared_ptr22#include <ostream>23#include <stdexcept>24#include <string>25#include <utility> // std::move26#include <vector>27 28namespace fair::mq {29 30/**31 * @class Channel Channel.h <fairmq/Channel.h>32 * @brief Wrapper class for Socket and related methods33 *34 * The class is not thread-safe.35 */36class Channel37{38 friend class Device;39 40 public:41 /// Default constructor42 Channel();43 44 /// Constructor45 /// @param name Channel name46 Channel(const std::string& name);47 48 /// Constructor49 /// @param type Socket type (push/pull/pub/sub/spub/xsub/pair/req/rep/dealer/router/)50 /// @param method Socket method (bind/connect)51 /// @param address Network address to bind/connect to (e.g. "tcp://127.0.0.1:5555" or "ipc://abc")52 Channel(const std::string& type, const std::string& method, const std::string& address);53 54 /// Constructor55 /// @param name Channel name56 /// @param type Socket type (push/pull/pub/sub/spub/xsub/pair/req/rep/dealer/router/)57 /// @param factory TransportFactory58 Channel(const std::string& name, const std::string& type, std::shared_ptr<TransportFactory> factory);59 60 /// Constructor61 /// @param name Channel name62 /// @param type Socket type (push/pull/pub/sub/spub/xsub/pair/req/rep/dealer/router/)63 /// @param method Socket method (bind/connect)64 /// @param address Network address to bind/connect to (e.g. "tcp://127.0.0.1:5555" or "ipc://abc")65 /// @param factory TransportFactory66 Channel(std::string name, std::string type, std::string method, std::string address, std::shared_ptr<TransportFactory> factory);67 68 Channel(const std::string& name, int index, const Properties& properties);69 70 /// Copy Constructor71 Channel(const Channel&);72 73 /// Copy Constructor (with new name)74 Channel(const Channel&, std::string name);75 76 /// Move constructor77 Channel(Channel&&) = default;78 79 /// Assignment operator80 Channel& operator=(const Channel&);81 82 /// Move assignment operator83 Channel& operator=(Channel&&) = default;84 85 /// Destructor86 ~Channel() = default;87 // { LOG(warn) << "Destroying channel '" << fName << "'"; }88 89 struct ChannelConfigurationError : std::runtime_error { using std::runtime_error::runtime_error; };90 91 Socket& GetSocket() const92 {93 assert(fSocket); // NOLINT(cppcoreguidelines-pro-bounds-array-to-pointer-decay)94 return *fSocket;95 }96 97 bool Bind(const std::string& address)98 {99 fMethod = "bind";100 fAddress = address;101 return fSocket->Bind(address);102 }103 104 bool Connect(const std::string& address)105 {106 fMethod = "connect";107 fAddress = address;108 return fSocket->Connect(address);109 }110 111 /// Get channel name112 /// @return Returns full channel name (e.g. "data[0]")113 std::string GetName() const { return fName; }114 115 /// Get channel prefix116 /// @return Returns channel prefix (e.g. "data" in "data[0]")117 std::string GetPrefix() const118 {119 std::string prefix = fName;120 prefix = prefix.erase(fName.rfind('['));121 return prefix;122 }123 124 /// Get channel index125 /// @return Returns channel index (e.g. 0 in "data[0]")126 std::string GetIndex() const127 {128 std::string indexStr = fName;129 indexStr.erase(indexStr.rfind(']'));130 indexStr.erase(0, indexStr.rfind('[') + 1);131 return indexStr;132 }133 134 /// Get socket type135 /// @return Returns socket type (push/pull/pub/sub/spub/xsub/pair/req/rep/dealer/router/)136 std::string GetType() const { return fType; }137 138 /// Get socket method139 /// @return Returns socket method (bind/connect)140 std::string GetMethod() const { return fMethod; }141 142 /// Get socket address (e.g. "tcp://127.0.0.1:5555" or "ipc://abc")143 /// @return Returns socket address (e.g. "tcp://127.0.0.1:5555" or "ipc://abc")144 std::string GetAddress() const { return fAddress; }145 146 /// Get channel transport name ("default", "zeromq" or "shmem")147 /// @return Returns channel transport name (e.g. "default", "zeromq" or "shmem")148 std::string GetTransportName() const;149 150 /// Get channel transport type151 /// @return Returns channel transport type152 mq::Transport GetTransportType() const;153 154 /// Get socket send buffer size (in number of messages)155 /// @return Returns socket send buffer size (in number of messages)156 int GetSndBufSize() const { return fSndBufSize; }157 158 /// Get socket receive buffer size (in number of messages)159 /// @return Returns socket receive buffer size (in number of messages)160 int GetRcvBufSize() const { return fRcvBufSize; }161 162 /// Get socket kernel transmit send buffer size (in bytes)163 /// @return Returns socket kernel transmit send buffer size (in bytes)164 int GetSndKernelSize() const { return fSndKernelSize; }165 166 /// Get socket kernel transmit receive buffer size (in bytes)167 /// @return Returns socket kernel transmit receive buffer size (in bytes)168 int GetRcvKernelSize() const { return fRcvKernelSize; }169 170 /// Get socket default send timeout (in ms)171 /// @return Returns socket default send timeout (in ms)172 int GetSndTimeout() const { return fSndTimeoutMs; }173 174 /// Get socket default receive timeout (in ms)175 /// @return Returns socket default receive timeout (in ms)176 int GetRcvTimeout() const { return fRcvTimeoutMs; }177 178 /// Get linger duration (in milliseconds)179 /// @return Returns linger duration (in milliseconds)180 int GetLinger() const { return fLinger; }181 182 /// Get socket rate logging interval (in seconds)183 /// @return Returns socket rate logging interval (in seconds)184 int GetRateLogging() const { return fRateLogging; }185 186 /// Get start of the port range for automatic binding187 /// @return start of the port range188 int GetPortRangeMin() const { return fPortRangeMin; }189 190 /// Get end of the port range for automatic binding191 /// @return end of the port range192 int GetPortRangeMax() const { return fPortRangeMax; }193 194 /// Set automatic binding (pick random port if bind fails)195 /// @return true/false, true if automatic binding is enabled196 bool GetAutoBind() const { return fAutoBind; }197 198 /// @par Thread Safety199 /// * @e Distinct @e objects: Safe.@n200 /// * @e Shared @e objects: Unsafe.201 auto GetNumberOfConnectedPeers() const202 {203 return fSocket ? fSocket->GetNumberOfConnectedPeers() : 0;204 }205 206 /// Set channel name207 /// @param name Arbitrary channel name208 void UpdateName(const std::string& name) { fName = name; Invalidate(); }209 210 /// Set socket type211 /// @param type Socket type (push/pull/pub/sub/spub/xsub/pair/req/rep/dealer/router/)212 void UpdateType(const std::string& type) { fType = type; Invalidate(); }213 214 /// Set socket method215 /// @param method Socket method (bind/connect)216 void UpdateMethod(const std::string& method) { fMethod = method; Invalidate(); }217 218 /// Set socket address219 /// @param Socket address (e.g. "tcp://127.0.0.1:5555" or "ipc://abc")220 void UpdateAddress(const std::string& address) { fAddress = address; Invalidate(); }221 222 /// Set channel transport223 /// @param transport transport string ("default", "zeromq" or "shmem")224 void UpdateTransport(const std::string& transport);225 226 /// Set socket send buffer size227 /// @param sndBufSize Socket send buffer size (in number of messages)228 void UpdateSndBufSize(int sndBufSize) { fSndBufSize = sndBufSize; Invalidate(); }229 230 /// Set socket receive buffer size231 /// @param rcvBufSize Socket receive buffer size (in number of messages)232 void UpdateRcvBufSize(int rcvBufSize) { fRcvBufSize = rcvBufSize; Invalidate(); }233 234 /// Set socket kernel transmit send buffer size (in bytes)235 /// @param sndKernelSize Socket send buffer size (in bytes)236 void UpdateSndKernelSize(int sndKernelSize) { fSndKernelSize = sndKernelSize; Invalidate(); }237 238 /// Set socket kernel transmit receive buffer size (in bytes)239 /// @param rcvKernelSize Socket receive buffer size (in bytes)240 void UpdateRcvKernelSize(int rcvKernelSize) { fRcvKernelSize = rcvKernelSize; Invalidate(); }241 242 /// Set socket default send timeout (in ms)243 /// @param sndTimeoutMs Socket default send timeout (in ms)244 void UpdateSndTimeout(int sndTimeoutMs) { fSndTimeoutMs = sndTimeoutMs; Invalidate(); }245 246 /// Set socket default receive timeout (in ms)247 /// @param rcvTimeoutMs Socket default receive timeout (in ms)248 void UpdateRcvTimeout(int rcvTimeoutMs) { fRcvTimeoutMs = rcvTimeoutMs; Invalidate(); }249 250 /// Set linger duration (in milliseconds)251 /// @param duration linger duration (in milliseconds)252 void UpdateLinger(int duration) { fLinger = duration; Invalidate(); }253 254 /// Set socket rate logging interval (in seconds)255 /// @param rateLogging Socket rate logging interval (in seconds)256 void UpdateRateLogging(int rateLogging) { fRateLogging = rateLogging; Invalidate(); }257 258 /// Set start of the port range for automatic binding259 /// @param minPort start of the port range260 void UpdatePortRangeMin(int minPort) { fPortRangeMin = minPort; Invalidate(); }261 262 /// Set end of the port range for automatic binding263 /// @param maxPort end of the port range264 void UpdatePortRangeMax(int maxPort) { fPortRangeMax = maxPort; Invalidate(); }265 266 /// Set automatic binding (pick random port if bind fails)267 /// @param autobind true/false, true to enable automatic binding268 void UpdateAutoBind(bool autobind) { fAutoBind = autobind; Invalidate(); }269 270 /// Checks if the configured channel settings are valid (checks the validity parameter, without running full validation (as oposed to ValidateChannel()))271 /// @return true if channel settings are valid, false otherwise.272 bool IsValid() const { return fValid; }273 274 /// Validates channel configuration275 /// @return true if channel settings are valid, false otherwise.276 bool Validate();277 278 void Init();279 280 bool ConnectEndpoint(const std::string& endpoint);281 282 bool BindEndpoint(std::string& endpoint);283 284 /// invalidates the channel (requires validation to be used again).285 void Invalidate() { fValid = false; }286 287 /// Send message(s) to the socket queue.288 /// @param m reference to MessagePtr/Parts/vector<MessagePtr>289 /// @param sndTimeoutMs send timeout in ms.290 /// -1 will wait forever (or until interrupt (e.g. via state change)),291 /// 0 will not wait (return immediately if cannot send).292 /// If not provided, default timeout will be taken.293 /// @return Number of bytes that have been queued,294 /// TransferCode::timeout if timed out,295 /// TransferCode::error if there was an error,296 /// TransferCode::interrupted if interrupted (e.g. by requested state change)297 template<typename M, typename... Timeout>298 std::enable_if_t<is_transferrable<M>::value, int64_t>299 Send(M& m, Timeout&&... sndTimeoutMs)300 {301 static_assert(sizeof...(sndTimeoutMs) <= 1, "Send called with too many arguments");302 303 CheckSendCompatibility(m);304 int t = fSndTimeoutMs;305 if constexpr (sizeof...(sndTimeoutMs) == 1) {306 t = {sndTimeoutMs...};307 }308 return fSocket->Send(m, t);309 }310 311 /// Receive message(s) from the socket queue.312 /// @param m reference to MessagePtr/Parts/vector<MessagePtr>313 /// @param rcvTimeoutMs receive timeout in ms.314 /// -1 will wait forever (or until interrupt (e.g. via state change)),315 /// 0 will not wait (return immediately if cannot receive).316 /// If not provided, default timeout will be taken.317 /// @return Number of bytes that have been received,318 /// TransferCode::timeout if timed out,319 /// TransferCode::error if there was an error,320 /// TransferCode::interrupted if interrupted (e.g. by requested state change)321 template<typename M, typename... Timeout>322 std::enable_if_t<is_transferrable<M>::value, int64_t>323 Receive(M& m, Timeout&&... rcvTimeoutMs)324 {325 static_assert(sizeof...(rcvTimeoutMs) <= 1, "Receive called with too many arguments");326 327 CheckReceiveCompatibility(m);328 int t = fRcvTimeoutMs;329 if constexpr (sizeof...(rcvTimeoutMs) == 1) {330 t = {rcvTimeoutMs...};331 }332 return fSocket->Receive(m, t);333 }334 335 unsigned long GetBytesTx() const { return fSocket->GetBytesTx(); }336 unsigned long GetBytesRx() const { return fSocket->GetBytesRx(); }337 unsigned long GetMessagesTx() const { return fSocket->GetMessagesTx(); }338 unsigned long GetMessagesRx() const { return fSocket->GetMessagesRx(); }339 340 auto Transport() -> TransportFactory* { return fTransportFactory.get(); };341 342 template<typename... Args>343 MessagePtr NewMessage(Args&&... args)344 {345 return Transport()->CreateMessage(std::forward<Args>(args)...);346 }347 348 template<typename T>349 MessagePtr NewSimpleMessage(const T& data)350 {351 return Transport()->NewSimpleMessage(data);352 }353 354 template<typename T>355 MessagePtr NewStaticMessage(const T& data)356 {357 return Transport()->NewStaticMessage(data);358 }359 360 template<typename... Args>361 UnmanagedRegionPtr NewUnmanagedRegion(Args&&... args)362 {363 return Transport()->CreateUnmanagedRegion(std::forward<Args>(args)...);364 }365 366 static constexpr mq::Transport DefaultTransportType = mq::Transport::DEFAULT;367 static constexpr const char* DefaultTransportName = "default";368 static constexpr const char* DefaultName = "";369 static constexpr const char* DefaultType = "unspecified";370 static constexpr const char* DefaultMethod = "unspecified";371 static constexpr const char* DefaultAddress = "unspecified";372 static constexpr int DefaultSndBufSize = 1000;373 static constexpr int DefaultRcvBufSize = 1000;374 static constexpr int DefaultSndKernelSize = 0;375 static constexpr int DefaultRcvKernelSize = 0;376 static constexpr int DefaultSndTimeoutMs = -1;377 static constexpr int DefaultRcvTimeoutMs = -1;378 static constexpr int DefaultLinger = 500;379 static constexpr int DefaultRateLogging = 1;380 static constexpr int DefaultPortRangeMin = 22000;381 static constexpr int DefaultPortRangeMax = 23000;382#ifdef FAIRMQ_CHANNEL_DEFAULT_AUTOBIND383 static constexpr bool DefaultAutoBind = FAIRMQ_CHANNEL_DEFAULT_AUTOBIND;384#else385 static constexpr bool DefaultAutoBind = true;386#endif387 388 friend std::ostream& operator<<(std::ostream& os, const Channel& ch)389 {390 return os << "name: " << ch.fName391 << ", type: " << ch.fType392 << ", method: " << ch.fMethod393 << ", address: " << ch.fAddress;394 }395 396 private:397 std::shared_ptr<TransportFactory> fTransportFactory;398 mq::Transport fTransportType;399 std::unique_ptr<Socket> fSocket;400 401 std::string fName;402 std::string fType;403 std::string fMethod;404 std::string fAddress;405 int fSndBufSize;406 int fRcvBufSize;407 int fSndKernelSize;408 int fRcvKernelSize;409 int fSndTimeoutMs;410 int fRcvTimeoutMs;411 int fLinger;412 int fRateLogging;413 int fPortRangeMin;414 int fPortRangeMax;415 bool fAutoBind;416 417 bool fValid;418 419 bool fMultipart;420 421 void CheckSendCompatibility(MessagePtr& msg)422 {423 if (fTransportType != msg->GetType()) {424 if (msg->GetSize() > 0) {425 MessagePtr msgWrapper(NewMessage(426 msg->GetData(),427 msg->GetSize(),428 [](void* /*data*/, void* _msg) { delete static_cast<Message*>(_msg); },429 msg.get()430 ));431 msg.release();432 msg = std::move(msgWrapper);433 } else {434 MessagePtr newMsg(NewMessage());435 msg = std::move(newMsg);436 }437 }438 }439 440 void CheckSendCompatibility(Parts& parts) { CheckSendCompatibility(parts.fParts); }441 void CheckSendCompatibility(Parts::container & msgVec)442 {443 for (auto& msg : msgVec) {444 if (fTransportType != msg->GetType()) {445 if (msg->GetSize() > 0) {446 MessagePtr msgWrapper(NewMessage(447 msg->GetData(),448 msg->GetSize(),449 [](void* /*data*/, void* _msg) { delete static_cast<Message*>(_msg); },450 msg.get()451 ));452 msg.release();453 msg = std::move(msgWrapper);454 } else {455 MessagePtr newMsg(NewMessage());456 msg = std::move(newMsg);457 }458 }459 }460 }461 462 void CheckReceiveCompatibility(MessagePtr& msg)463 {464 if (fTransportType != msg->GetType()) {465 MessagePtr newMsg(NewMessage());466 msg = std::move(newMsg);467 }468 }469 470 void CheckReceiveCompatibility(Parts& parts) { CheckReceiveCompatibility(parts.fParts); }471 void CheckReceiveCompatibility(Parts::container& msgVec)472 {473 for (auto& msg : msgVec) {474 if (fTransportType != msg->GetType()) {475 476 MessagePtr newMsg(NewMessage());477 msg = std::move(newMsg);478 }479 }480 }481 482 void InitTransport(std::shared_ptr<TransportFactory> factory)483 {484 fTransportFactory = factory;485 fTransportType = factory->GetType();486 }487};488 489} // namespace fair::mq490 491using FairMQChannel [[deprecated("Use fair::mq::Channel")]] = fair::mq::Channel;492 493#endif // FAIR_MQ_CHANNEL_H