Source · fairmq/
Channel.cxxsource
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) 2014-2021 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#include <boost/algorithm/string.hpp> // join/split10#include <cstddef> // size_t11#include <fairlogger/Logger.h>12#include <fairmq/Channel.h>13#include <fairmq/Properties.h>14#include <fairmq/Tools.h>15#include <fairmq/Transports.h>16#include <random>17#include <regex>18#include <set>19 20namespace fair::mq {21 22using namespace std;23 24template<typename T>25T GetPropertyOrDefault(const Properties& m, const string& k, const T& ifNotFound)26{27 if (m.count(k)) {28 return boost::any_cast<T>(m.at(k));29 }30 return ifNotFound;31}32 33constexpr Transport Channel::DefaultTransportType;34constexpr const char* Channel::DefaultTransportName;35constexpr const char* Channel::DefaultName;36constexpr const char* Channel::DefaultType;37constexpr const char* Channel::DefaultMethod;38constexpr const char* Channel::DefaultAddress;39constexpr int Channel::DefaultSndBufSize;40constexpr int Channel::DefaultRcvBufSize;41constexpr int Channel::DefaultSndKernelSize;42constexpr int Channel::DefaultRcvKernelSize;43constexpr int Channel::DefaultSndTimeoutMs;44constexpr int Channel::DefaultRcvTimeoutMs;45constexpr int Channel::DefaultLinger;46constexpr int Channel::DefaultRateLogging;47constexpr int Channel::DefaultPortRangeMin;48constexpr int Channel::DefaultPortRangeMax;49constexpr bool Channel::DefaultAutoBind;50 51Channel::Channel()52 : Channel(DefaultName, DefaultType, DefaultMethod, DefaultAddress, nullptr)53{}54 55Channel::Channel(const string& name)56 : Channel(name, DefaultType, DefaultMethod, DefaultAddress, nullptr)57{}58 59Channel::Channel(const string& type, const string& method, const string& address)60 : Channel(DefaultName, type, method, address, nullptr)61{}62 63Channel::Channel(const string& name, const string& type, shared_ptr<TransportFactory> factory)64 : Channel(name, type, DefaultMethod, DefaultAddress, factory)65{}66 67Channel::Channel(string name, string type, string method, string address, shared_ptr<TransportFactory> factory)68 : fTransportFactory(factory)69 , fTransportType(factory ? factory->GetType() : DefaultTransportType)70 , fSocket(factory ? factory->CreateSocket(type, name) : nullptr)71 , fName(std::move(name))72 , fType(std::move(type))73 , fMethod(std::move(method))74 , fAddress(std::move(address))75 , fSndBufSize(DefaultSndBufSize)76 , fRcvBufSize(DefaultRcvBufSize)77 , fSndKernelSize(DefaultSndKernelSize)78 , fRcvKernelSize(DefaultRcvKernelSize)79 , fSndTimeoutMs(DefaultSndTimeoutMs)80 , fRcvTimeoutMs(DefaultRcvTimeoutMs)81 , fLinger(DefaultLinger)82 , fRateLogging(DefaultRateLogging)83 , fPortRangeMin(DefaultPortRangeMin)84 , fPortRangeMax(DefaultPortRangeMax)85 , fAutoBind(DefaultAutoBind)86 , fValid(false)87 , fMultipart(false)88{89 // LOG(warn) << "Constructing channel '" << fName << "'";90}91 92Channel::Channel(const string& name, int index, const Properties& properties)93 : Channel(tools::ToString(name, "[", index, "]"), "unspecified", "unspecified", "unspecified", nullptr)94{95 string prefix(tools::ToString("chans.", name, ".", index, "."));96 97 fType = GetPropertyOrDefault(properties, string(prefix + "type"), std::string(DefaultType));98 fMethod = GetPropertyOrDefault(properties, string(prefix + "method"), std::string(DefaultMethod));99 fAddress = GetPropertyOrDefault(properties, string(prefix + "address"), std::string(DefaultAddress));100 fTransportType = TransportType(GetPropertyOrDefault(properties, string(prefix + "transport"), std::string(DefaultTransportName)));101 fSndBufSize = GetPropertyOrDefault(properties, string(prefix + "sndBufSize"), DefaultSndBufSize);102 fRcvBufSize = GetPropertyOrDefault(properties, string(prefix + "rcvBufSize"), DefaultRcvBufSize);103 fSndKernelSize = GetPropertyOrDefault(properties, string(prefix + "sndKernelSize"), DefaultSndKernelSize);104 fRcvKernelSize = GetPropertyOrDefault(properties, string(prefix + "rcvKernelSize"), DefaultRcvKernelSize);105 fSndTimeoutMs = GetPropertyOrDefault(properties, string(prefix + "sndTimeoutMs"), DefaultSndTimeoutMs);106 fRcvTimeoutMs = GetPropertyOrDefault(properties, string(prefix + "rcvTimeoutMs"), DefaultRcvTimeoutMs);107 fLinger = GetPropertyOrDefault(properties, string(prefix + "linger"), DefaultLinger);108 fRateLogging = GetPropertyOrDefault(properties, string(prefix + "rateLogging"), DefaultRateLogging);109 fPortRangeMin = GetPropertyOrDefault(properties, string(prefix + "portRangeMin"), DefaultPortRangeMin);110 fPortRangeMax = GetPropertyOrDefault(properties, string(prefix + "portRangeMax"), DefaultPortRangeMax);111 fAutoBind = GetPropertyOrDefault(properties, string(prefix + "autoBind"), DefaultAutoBind);112}113 114Channel::Channel(const Channel& chan)115 : Channel(chan, chan.fName)116{}117 118Channel::Channel(const Channel& chan, string newName)119 : fTransportFactory(nullptr)120 , fTransportType(chan.fTransportType)121 , fSocket(nullptr)122 , fName(std::move(newName))123 , fType(chan.fType)124 , fMethod(chan.fMethod)125 , fAddress(chan.fAddress)126 , fSndBufSize(chan.fSndBufSize)127 , fRcvBufSize(chan.fRcvBufSize)128 , fSndKernelSize(chan.fSndKernelSize)129 , fRcvKernelSize(chan.fRcvKernelSize)130 , fSndTimeoutMs(chan.fSndTimeoutMs)131 , fRcvTimeoutMs(chan.fRcvTimeoutMs)132 , fLinger(chan.fLinger)133 , fRateLogging(chan.fRateLogging)134 , fPortRangeMin(chan.fPortRangeMin)135 , fPortRangeMax(chan.fPortRangeMax)136 , fAutoBind(chan.fAutoBind)137 , fValid(false)138 , fMultipart(chan.fMultipart)139{}140 141Channel& Channel::operator=(const Channel& chan)142{143 if (this == &chan) {144 return *this;145 }146 147 fTransportFactory = nullptr;148 fTransportType = chan.fTransportType;149 fSocket = nullptr;150 fName = chan.fName;151 fType = chan.fType;152 fMethod = chan.fMethod;153 fAddress = chan.fAddress;154 fSndBufSize = chan.fSndBufSize;155 fRcvBufSize = chan.fRcvBufSize;156 fSndKernelSize = chan.fSndKernelSize;157 fRcvKernelSize = chan.fRcvKernelSize;158 fSndTimeoutMs = chan.fSndTimeoutMs;159 fRcvTimeoutMs = chan.fRcvTimeoutMs;160 fLinger = chan.fLinger;161 fRateLogging = chan.fRateLogging;162 fPortRangeMin = chan.fPortRangeMin;163 fPortRangeMax = chan.fPortRangeMax;164 fAutoBind = chan.fAutoBind;165 fValid = false;166 fMultipart = chan.fMultipart;167 168 return *this;169}170 171bool Channel::Validate()172try {173 stringstream ss;174 ss << "Validating channel '" << fName << "'... ";175 176 if (fValid) {177 ss << "ALREADY VALID";178 LOG(debug) << ss.str();179 return true;180 }181 182 // validate channel name183 smatch m;184 static regex const invalidName(R"([^a-zA-Z0-9\-_\[\]#])");185 if (regex_search(fName, m, invalidName)) {186 ss << "INVALID";187 LOG(debug) << ss.str();188 LOG(error) << "channel name contains illegal character: '" << m.str(0) << "', allowed characters are: a-z, A-Z, 0-9, -, _, [, ], #";189 return false;190 }191 192 // validate socket type193 const set<string> socketTypes{ "sub", "pub", "pull", "push", "req", "rep", "xsub", "xpub", "dealer", "router", "pair" };194 if (socketTypes.find(fType) == socketTypes.end()) {195 ss << "INVALID";196 LOG(debug) << ss.str();197 LOG(error) << "Invalid channel type: '" << fType << "'";198 throw ChannelConfigurationError(tools::ToString("Invalid channel type: '", fType, "'"));199 }200 201 // validate socket address202 if (fAddress == "unspecified" || fAddress.empty()) {203 ss << "INVALID";204 LOG(debug) << ss.str();205 LOG(debug) << "invalid channel address: '" << fAddress << "'";206 return false;207 } else {208 vector<string> endpoints;209 boost::algorithm::split(endpoints, fAddress, boost::algorithm::is_any_of(";"));210 for (const auto& endpoint : endpoints) {211 string address;212 if (endpoint[0] == '@' || endpoint[0] == '+' || endpoint[0] == '>') {213 address = endpoint.substr(1);214 } else {215 // we don't have a method modifier, check if the default method is set216 const set<string> socketMethods{ "bind", "connect" };217 if (socketMethods.find(fMethod) == socketMethods.end()) {218 ss << "INVALID";219 LOG(debug) << ss.str();220 LOG(error) << "Invalid endpoint connection method: '" << fMethod << "' for " << endpoint;221 throw ChannelConfigurationError(tools::ToString("Invalid endpoint connection method: '", fMethod, "' for ", endpoint));222 }223 address = endpoint;224 }225 // check if address is a tcp or ipc address226 if (address.compare(0, 6, "tcp://") == 0) {227 // check if TCP address contains port delimiter228 string addressString = address.substr(6);229 if (addressString.find(':') == string::npos) {230 ss << "INVALID";231 LOG(debug) << ss.str();232 LOG(error) << "invalid channel address: '" << address << "' (missing port?)";233 return false;234 }235 } else if (address.compare(0, 6, "ipc://") == 0) {236 // check if IPC address is not empty237 string addressString = address.substr(6);238 if (addressString.empty()) {239 ss << "INVALID";240 LOG(debug) << ss.str();241 LOG(error) << "invalid channel address: '" << address << "' (empty IPC address?)";242 return false;243 }244 } else if (address.compare(0, 9, "inproc://") == 0) {245 // check if IPC address is not empty246 string addressString = address.substr(9);247 if (addressString.empty()) {248 ss << "INVALID";249 LOG(debug) << ss.str();250 LOG(error) << "invalid channel address: '" << address << "' (empty inproc address?)";251 return false;252 }253 } else if (address.compare(0, 8, "verbs://") == 0) {254 // check if IPC address is not empty255 string addressString = address.substr(8);256 if (addressString.empty()) {257 ss << "INVALID";258 LOG(debug) << ss.str();259 LOG(error) << "invalid channel address: '" << address << "' (empty verbs address?)";260 return false;261 }262 } else {263 // if neither TCP or IPC is specified, return invalid264 ss << "INVALID";265 LOG(debug) << ss.str();266 LOG(error) << "invalid channel address: '" << address << "' (missing protocol specifier?)";267 return false;268 }269 }270 }271 272 // validate socket buffer size for sending273 if (fSndBufSize < 0) {274 ss << "INVALID";275 LOG(debug) << ss.str();276 LOG(error) << "invalid channel send buffer size (cannot be negative): '" << fSndBufSize << "'";277 throw ChannelConfigurationError(tools::ToString("invalid channel send buffer size (cannot be negative): '", fSndBufSize, "'"));278 }279 280 // validate socket buffer size for receiving281 if (fRcvBufSize < 0) {282 ss << "INVALID";283 LOG(debug) << ss.str();284 LOG(error) << "invalid channel receive buffer size (cannot be negative): '" << fRcvBufSize << "'";285 throw ChannelConfigurationError(tools::ToString("invalid channel receive buffer size (cannot be negative): '", fRcvBufSize, "'"));286 }287 288 // validate socket kernel transmit size for sending289 if (fSndKernelSize < 0) {290 ss << "INVALID";291 LOG(debug) << ss.str();292 LOG(error) << "invalid channel send kernel transmit size (cannot be negative): '" << fSndKernelSize << "'";293 throw ChannelConfigurationError(tools::ToString("invalid channel send kernel transmit size (cannot be negative): '", fSndKernelSize, "'"));294 }295 296 // validate socket kernel transmit size for receiving297 if (fRcvKernelSize < 0) {298 ss << "INVALID";299 LOG(debug) << ss.str();300 LOG(error) << "invalid channel receive kernel transmit size (cannot be negative): '" << fRcvKernelSize << "'";301 throw ChannelConfigurationError(tools::ToString("invalid channel receive kernel transmit size (cannot be negative): '", fRcvKernelSize, "'"));302 }303 304 // validate socket rate logging interval305 if (fRateLogging < 0) {306 ss << "INVALID";307 LOG(debug) << ss.str();308 LOG(error) << "invalid socket rate logging interval (cannot be negative): '" << fRateLogging << "'";309 throw ChannelConfigurationError(tools::ToString("invalid socket rate logging interval (cannot be negative): '", fRateLogging, "'"));310 }311 312 fValid = true;313 ss << "VALID";314 LOG(debug) << ss.str();315 return true;316} catch (exception& e) {317 LOG(error) << "Exception caught in Channel::ValidateChannel: " << e.what();318 throw ChannelConfigurationError(tools::ToString(e.what()));319}320 321void Channel::Init()322{323 fSocket = fTransportFactory->CreateSocket(fType, fName);324 325 // set linger duration (how long socket should wait for outstanding transfers before shutdown)326 fSocket->SetLinger(fLinger);327 328 // set high water marks329 fSocket->SetSndBufSize(fSndBufSize);330 fSocket->SetRcvBufSize(fRcvBufSize);331 332 // set kernel transmit size (set it only if value is not the default value)333 if (fSndKernelSize != 0) {334 fSocket->SetSndKernelSize(fSndKernelSize);335 }336 if (fRcvKernelSize != 0) {337 fSocket->SetRcvKernelSize(fRcvKernelSize);338 }339}340 341bool Channel::ConnectEndpoint(const string& endpoint)342{343 return fSocket->Connect(endpoint);344}345 346bool Channel::BindEndpoint(string& endpoint)347{348 // try to bind to the configured port. If it fails, try random one (if AutoBind is on).349 if (fSocket->Bind(endpoint)) {350 return true;351 } else {352 // auto-bind only implemented for TCP353 size_t protocolPos = endpoint.find(':');354 string protocol = endpoint.substr(0, protocolPos);355 if (protocol != "tcp") {356 return false;357 }358 359 if (fAutoBind) {360 // number of attempts when choosing a random port361 int numAttempts = 0;362 int maxAttempts = 1000;363 364 // initialize random generator365 default_random_engine generator(chrono::system_clock::now().time_since_epoch().count());366 uniform_int_distribution<int> randomPort(fPortRangeMin, fPortRangeMax);367 368 do {369 LOG(debug) << "Could not bind to configured (TCP) port (" << endpoint << "), trying random port in range " << fPortRangeMin << "-" << fPortRangeMax;370 ++numAttempts;371 372 if (numAttempts > maxAttempts) {373 LOG(error) << "could not bind to any (TCP) port in the given range after " << maxAttempts << " attempts";374 return false;375 }376 377 size_t pos = endpoint.rfind(':');378 endpoint = endpoint.substr(0, pos + 1) + tools::ToString(static_cast<int>(randomPort(generator)));379 } while (!fSocket->Bind(endpoint));380 381 return true;382 } else {383 return false;384 }385 }386}387 388std::string Channel::GetTransportName() const { return TransportName(fTransportType); }389 390Transport Channel::GetTransportType() const { return fTransportType; }391 392void Channel::UpdateTransport(const std::string& transport) { fTransportType = TransportType(transport); Invalidate(); }393 394} // namespace fair::mq