Source · fairmq/zeromq/
Common.hheader
Described in Transfer codes and errors.
1/********************************************************************************2 * Copyright (C) 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#ifndef FAIR_MQ_ZMQ_COMMON_H9#define FAIR_MQ_ZMQ_COMMON_H10 11#include <fairlogger/Logger.h>12#include <fairmq/Error.h>13#include <fairmq/tools/Strings.h>14#include <stdexcept>15#include <string_view>16#include <zmq.h>17 18namespace fair::mq::zmq19{20 21struct Error : std::runtime_error { using std::runtime_error::runtime_error; };22 23inline bool Bind(void* socket, const std::string& address, const std::string& id)24{25 // LOG(debug) << "Binding socket " << id << " on " << address;26 if (zmq_bind(socket, address.c_str()) != 0) {27 if (errno == EADDRINUSE) {28 // do not print error in this case, this is handled upstream in case no29 // connection could be established after trying a number of random ports from a range.30 size_t protocolPos = address.find(':');31 std::string protocol = address.substr(0, protocolPos);32 if (protocol == "tcp") {33 return false;34 }35 } else if (errno == EACCES) {36 // check if TCP port 1 was given, if yes then it will be handeled upstream, print debug only37 size_t protocolPos = address.find(':');38 std::string protocol = address.substr(0, protocolPos);39 if (protocol == "tcp") {40 size_t portPos = address.rfind(':');41 std::string port = address.substr(portPos + 1);42 if (port == "1") {43 LOG(debug) << "Failed binding socket " << id << ", address: " << address << ", reason: " << zmq_strerror(errno);44 return false;45 }46 }47 }48 LOG(error) << "Failed binding socket " << id << ", address: " << address << ", reason: " << zmq_strerror(errno);49 return false;50 }51 return true;52}53 54inline bool Connect(void* socket, const std::string& address, const std::string& id)55{56 // LOG(debug) << "Connecting socket " << id << " on " << address;57 if (zmq_connect(socket, address.c_str()) != 0) {58 LOG(error) << "Failed connecting socket " << id << ", address: " << address << ", reason: " << zmq_strerror(errno);59 return false;60 }61 return true;62}63 64inline bool ShouldRetry(int flags, int socketTimeout, int userTimeout, int& elapsed)65{66 if ((flags & ZMQ_DONTWAIT) == 0) {67 if (userTimeout > 0) {68 elapsed += socketTimeout;69 if (elapsed >= userTimeout) {70 return false;71 }72 }73 return true;74 } else {75 return false;76 }77}78 79inline int HandleErrors(const std::string& id)80{81 if (zmq_errno() == ETERM) {82 LOG(debug) << "Terminating socket " << id;83 return static_cast<int>(TransferCode::error);84 } else {85 LOG(error) << "Failed transfer on socket " << id << ", errno: " << errno << ", reason: " << zmq_strerror(errno);86 return static_cast<int>(TransferCode::error);87 }88}89 90/// Lookup table for various zmq constants91inline auto getConstant(std::string_view constant) -> int92{93 if (constant.empty()) { return 0; }94 if (constant == "sub") { return ZMQ_SUB; }95 if (constant == "pub") { return ZMQ_PUB; }96 if (constant == "xsub") { return ZMQ_XSUB; }97 if (constant == "xpub") { return ZMQ_XPUB; }98 if (constant == "push") { return ZMQ_PUSH; }99 if (constant == "pull") { return ZMQ_PULL; }100 if (constant == "req") { return ZMQ_REQ; }101 if (constant == "rep") { return ZMQ_REP; }102 if (constant == "dealer") { return ZMQ_DEALER; }103 if (constant == "router") { return ZMQ_ROUTER; }104 if (constant == "pair") { return ZMQ_PAIR; }105 106 if (constant == "snd-hwm") { return ZMQ_SNDHWM; }107 if (constant == "rcv-hwm") { return ZMQ_RCVHWM; }108 if (constant == "snd-size") { return ZMQ_SNDBUF; }109 if (constant == "rcv-size") { return ZMQ_RCVBUF; }110 if (constant == "snd-more") { return ZMQ_SNDMORE; }111 if (constant == "rcv-more") { return ZMQ_RCVMORE; }112 113 if (constant == "linger") { return ZMQ_LINGER; }114 if (constant == "no-block") { return ZMQ_DONTWAIT; }115 if (constant == "snd-more no-block") { return ZMQ_DONTWAIT|ZMQ_SNDMORE; }116 117 if (constant == "fd") { return ZMQ_FD; }118 if (constant == "events") { return ZMQ_EVENTS; }119 if (constant == "pollin") { return ZMQ_POLLIN; }120 if (constant == "pollout") { return ZMQ_POLLOUT; }121 122 throw Error(tools::ToString("getConstant called with an invalid argument: ", constant));123}124 125/// Create a zmq event monitor socket pair, and configure/connect the reading socket126/// @return reading monitor socket127inline auto makeMonitorSocket(void* zmqCtx, void* socketToMonitor, std::string_view id) -> void*128{129 assertm(zmqCtx, "Given zmq context exists"); // NOLINT130 131 if (!socketToMonitor) { // nothing to do in this case132 return nullptr;133 }134 135 auto const address(tools::ToString("inproc://", id));136 { // Create writing monitor socket on socket to be monitored and subscribe137 // to all relevant events needed to compute connected peers138 // from http://api.zeromq.org/master:zmq-socket-monitor:139 //140 // ZMQ_EVENT_CONNECTED - The socket has successfully connected to a remote peer. The event141 // value is the file descriptor (FD) of the underlying network socket. Warning: there is142 // no guarantee that the FD is still valid by the time your code receives this event.143 // ZMQ_EVENT_ACCEPTED - The socket has accepted a connection from a remote peer. The event144 // value is the FD of the underlying network socket. Warning: there is no guarantee that145 // the FD is still valid by the time your code receives this event.146 // ZMQ_EVENT_DISCONNECTED - The socket was disconnected unexpectedly. The event value is the147 // FD of the underlying network socket. Warning: this socket will be closed.148 [[maybe_unused]] auto const rc =149 zmq_socket_monitor(socketToMonitor,150 address.c_str(),151 ZMQ_EVENT_CONNECTED | ZMQ_EVENT_ACCEPTED | ZMQ_EVENT_DISCONNECTED);152 assertm(rc == 0, "Creating writing monitor socket succeeded"); // NOLINT153 }154 // Create reading monitor socket155 auto mon(zmq_socket(zmqCtx, ZMQ_PAIR));156 assertm(mon, "Creating reading monitor socker succeeded"); // NOLINT157 { // Set receive queue size to unlimited on reading socket158 // Rationale: In the current implementation this is needed for correctness, because159 // we do not have any thread that emptys the receive queue regularly.160 // Progress only happens, when a user calls GetNumberOfConnectedPeers()`.161 // The assumption here is, that not too many events will pile up anyways.162 int const unlimited(0);163 [[maybe_unused]] auto const rc = zmq_setsockopt(mon, ZMQ_RCVHWM, &unlimited, sizeof(unlimited));164 assertm(rc == 0, "Setting rcv queue size to unlimited succeeded"); // NOLINT165 }166 { // Connect the reading monitor socket167 [[maybe_unused]] auto const rc = zmq_connect(mon, address.c_str());168 assertm(rc == 0, "Connecting reading monitor socket succeeded"); // NOLINT169 }170 return mon;171}172 173/// Stop monitoring, discard all pending events and close the reading monitor socket174///175/// Stopping the monitor first closes the writing end synchronously. Otherwise, events emitted176/// while the monitored socket is closing can be dropped by libzmq after the reading end is gone,177/// which leaks their (heap allocated) endpoint frames.178inline auto closeMonitorSocket(void* socketToMonitor, void* monitorSocket) -> void179{180 if (monitorSocket == nullptr) {181 return;182 }183 184 if (socketToMonitor != nullptr) {185 [[maybe_unused]] auto const rc = zmq_socket_monitor(socketToMonitor, nullptr, 0);186 }187 188 zmq_msg_t msg;189 zmq_msg_init(&msg);190 while (zmq_msg_recv(&msg, monitorSocket, ZMQ_DONTWAIT) != -1) {}191 zmq_msg_close(&msg);192 193 if (zmq_close(monitorSocket) != 0) {194 LOG(error) << "Failed closing monitor socket, reason: " << zmq_strerror(errno);195 }196}197 198/// Read pending zmq monitor event in a non-blocking fashion.199/// @return event id or -1 for no event pending200inline auto getMonitorEvent(void* monitorSocket) -> int201{202 assertm(monitorSocket, "zmq monitor socket exists"); // NOLINT203 204 // First frame in message contains event id205 zmq_msg_t msg;206 zmq_msg_init(&msg);207 {208 auto const size = zmq_msg_recv(&msg, monitorSocket, ZMQ_DONTWAIT);209 if (size == -1) {210 return -1; // no event pending211 }212 assertm(size >= 2, "At least two bytes were received"); // NOLINT213 }214 215 // Unpack event id216 auto const event = *static_cast<uint16_t*>(zmq_msg_data(&msg));217 218 // No unpacking of the event value needed for now219 220 // Second frame in message contains event address221 assertm(zmq_msg_more(&msg), "A second frame is pending"); // NOLINT222 zmq_msg_close(&msg);223 zmq_msg_t msg2;224 zmq_msg_init(&msg2);225 {226 [[maybe_unused]] auto const rc = zmq_msg_recv(&msg2, monitorSocket, 0);227 assertm(rc >= 0, "second monitor event frame successfully received"); // NOLINT228 }229 assertm(!zmq_msg_more(&msg2), "No more frames are pending"); // NOLINT230 // No unpacking of the event address needed for now231 zmq_msg_close(&msg2);232 233 return event;234}235 236/// Compute updated connected peers count by consuming pending events from a zmq monitor socket237/// @return updated connected peers count238inline auto updateNumberOfConnectedPeers(unsigned long count, void* monitorSocket) -> unsigned long239{240 if (monitorSocket == nullptr) {241 return count;242 }243 244 int event = getMonitorEvent(monitorSocket);245 while (event >= 0) {246 switch (event) {247 case ZMQ_EVENT_CONNECTED:248 case ZMQ_EVENT_ACCEPTED:249 ++count;250 break;251 case ZMQ_EVENT_DISCONNECTED:252 if (count > 0) {253 --count;254 } else {255 LOG(warn) << "Computing connected peers would result in negative count! Some event was missed!";256 }257 break;258 default:259 break;260 }261 event = getMonitorEvent(monitorSocket);262 }263 return count;264}265 266} // namespace fair::mq::zmq267 268#endif /* FAIR_MQ_ZMQ_COMMON_H */