Source · fairmq/zeromq/
Socket.hheader
Described in Channels and sockets, Channel properties.
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#ifndef FAIR_MQ_ZMQ_SOCKET_H10#define FAIR_MQ_ZMQ_SOCKET_H11 12#include <fairmq/Message.h>13#include <fairmq/Socket.h>14#include <fairmq/tools/Strings.h>15#include <fairmq/zeromq/Common.h>16#include <fairmq/zeromq/Context.h>17#include <fairmq/zeromq/Message.h>18 19#include <fairlogger/Logger.h>20 21#include <zmq.h>22 23#include <atomic>24#include <functional>25#include <memory> // unique_ptr, make_unique26#include <string_view>27 28namespace fair::mq::zmq29{30 31class Socket final : public fair::mq::Socket32{33 public:34 Socket(Context& ctx, const std::string& type, const std::string& name, const std::string& id, fair::mq::TransportFactory* factory = nullptr)35 : fair::mq::Socket(factory)36 , fCtx(ctx)37 , fId(id + "." + name + "." + type)38 , fSocket(zmq_socket(fCtx.GetZmqCtx(), getConstant(type)))39 , fMonitorSocket(makeMonitorSocket(fCtx.GetZmqCtx(), fSocket, fId))40 , fBytesTx(0)41 , fBytesRx(0)42 , fMessagesTx(0)43 , fMessagesRx(0)44 45 {46 if (fSocket == nullptr) {47 LOG(error) << "Failed creating socket " << fId << ", reason: " << zmq_strerror(errno);48 throw SocketError(tools::ToString("Unavailable transport requested: ", type));49 }50 51 if (zmq_setsockopt(fSocket, ZMQ_IDENTITY, fId.c_str(), fId.length()) != 0) {52 LOG(error) << "Failed setting ZMQ_IDENTITY socket option, reason: " << zmq_strerror(errno);53 }54 55 // Tell socket to try and send/receive outstanding messages for <linger> milliseconds before56 // terminating. Default value for ZeroMQ is -1, which is to wait forever.57 int linger = 1000;58 if (zmq_setsockopt(fSocket, ZMQ_LINGER, &linger, sizeof(linger)) != 0) {59 LOG(error) << "Failed setting ZMQ_LINGER socket option, reason: " << zmq_strerror(errno);60 }61 62 if (zmq_setsockopt(fSocket, ZMQ_SNDTIMEO, &fTimeout, sizeof(fTimeout)) != 0) {63 LOG(error) << "Failed setting ZMQ_SNDTIMEO socket option, reason: " << zmq_strerror(errno);64 }65 66 if (zmq_setsockopt(fSocket, ZMQ_RCVTIMEO, &fTimeout, sizeof(fTimeout)) != 0) {67 LOG(error) << "Failed setting ZMQ_RCVTIMEO socket option, reason: " << zmq_strerror(errno);68 }69 70 if (type == "sub") {71 if (zmq_setsockopt(fSocket, ZMQ_SUBSCRIBE, nullptr, 0) != 0) {72 LOG(error) << "Failed setting ZMQ_SUBSCRIBE socket option, reason: " << zmq_strerror(errno);73 }74 }75 76 LOG(debug) << "Created socket " << GetId();77 }78 79 Socket(const Socket&) = delete;80 Socket(Socket&&) = delete;81 Socket& operator=(const Socket&) = delete;82 Socket& operator=(Socket&&) = delete;83 84 std::string GetId() const override { return fId; }85 86 bool Bind(const std::string& address) override87 {88 return zmq::Bind(fSocket, address, fId);89 }90 91 bool Connect(const std::string& address) override92 {93 return zmq::Connect(fSocket, address, fId);94 }95 96 int64_t Send(MessagePtr& msg, int timeout = -1) override97 {98 int flags = 0;99 if (timeout == 0) {100 flags = ZMQ_DONTWAIT;101 }102 int elapsed = 0;103 104 int64_t actualBytes = zmq_msg_size(static_cast<Message*>(msg.get())->GetMessage());105 106 while (true) {107 int nbytes = zmq_msg_send(static_cast<Message*>(msg.get())->GetMessage(), fSocket, flags);108 if (nbytes >= 0) {109 fBytesTx += actualBytes;110 ++fMessagesTx;111 return actualBytes;112 } else if (zmq_errno() == EAGAIN || zmq_errno() == EINTR) {113 if (fCtx.Interrupted()) {114 return static_cast<int>(TransferCode::interrupted);115 } else if (zmq::ShouldRetry(flags, fTimeout, timeout, elapsed)) {116 continue;117 } else {118 return static_cast<int>(TransferCode::timeout);119 }120 } else {121 return zmq::HandleErrors(fId);122 }123 }124 }125 126 int64_t Receive(MessagePtr& msg, int timeout = -1) override127 {128 int flags = 0;129 if (timeout == 0) {130 flags = ZMQ_DONTWAIT;131 }132 int elapsed = 0;133 134 while (true) {135 int nbytes = zmq_msg_recv(static_cast<Message*>(msg.get())->GetMessage(), fSocket, flags);136 if (nbytes >= 0) {137 static_cast<Message*>(msg.get())->Realign();138 int64_t actualBytes = zmq_msg_size(static_cast<Message*>(msg.get())->GetMessage());139 fBytesRx += actualBytes;140 ++fMessagesRx;141 return actualBytes;142 } else if (zmq_errno() == EAGAIN || zmq_errno() == EINTR) {143 if (fCtx.Interrupted()) {144 return static_cast<int>(TransferCode::interrupted);145 } else if (zmq::ShouldRetry(flags, fTimeout, timeout, elapsed)) {146 continue;147 } else {148 return static_cast<int>(TransferCode::timeout);149 }150 } else {151 return zmq::HandleErrors(fId);152 }153 }154 }155 156 int64_t Send(Parts::container& msgVec, int timeout = -1) override157 {158 int flags = 0;159 if (timeout == 0) {160 flags = ZMQ_DONTWAIT;161 }162 163 const unsigned int vecSize = msgVec.size();164 165 // Sending vector typicaly handles more then one part166 if (vecSize > 1) {167 int elapsed = 0;168 169 while (true) {170 int64_t totalSize = 0;171 bool repeat = false;172 173 for (unsigned int i = 0; i < vecSize; ++i) {174 int nbytes = zmq_msg_send(static_cast<Message*>(msgVec[i].get())->GetMessage(), fSocket, (i < vecSize - 1) ? ZMQ_SNDMORE | flags : flags);175 if (nbytes >= 0) {176 totalSize += nbytes;177 } else if (zmq_errno() == EAGAIN || zmq_errno() == EINTR) {178 if (fCtx.Interrupted()) {179 return static_cast<int>(TransferCode::interrupted);180 } else if (zmq::ShouldRetry(flags, fTimeout, timeout, elapsed)) {181 repeat = true;182 break;183 } else {184 return static_cast<int>(TransferCode::timeout);185 }186 } else {187 return zmq::HandleErrors(fId);188 }189 }190 191 if (repeat) {192 continue;193 }194 195 // store statistics on how many messages have been sent (handle all parts as a single message)196 ++fMessagesTx;197 fBytesTx += totalSize;198 return totalSize;199 }200 } else if (vecSize == 1) { // If there's only one part, send it as a regular message201 return Send(msgVec.back(), timeout);202 } else { // if the vector is empty, something might be wrong203 LOG(warn) << "Will not send empty vector";204 return static_cast<int>(TransferCode::error);205 }206 }207 208 int64_t Receive(Parts::container& msgVec, int timeout = -1) override209 {210 int flags = 0;211 if (timeout == 0) {212 flags = ZMQ_DONTWAIT;213 }214 int elapsed = 0;215 216 while (true) {217 int64_t totalSize = 0;218 int more = 0;219 bool repeat = false;220 221 do {222 fair::mq::MessagePtr part = std::make_unique<Message>(GetTransport());223 224 int nbytes = zmq_msg_recv(static_cast<Message*>(part.get())->GetMessage(), fSocket, flags);225 if (nbytes >= 0) {226 static_cast<Message*>(part.get())->Realign();227 msgVec.push_back(std::move(part));228 totalSize += nbytes;229 } else if (zmq_errno() == EAGAIN || zmq_errno() == EINTR) {230 if (fCtx.Interrupted()) {231 return static_cast<int>(TransferCode::interrupted);232 } else if (zmq::ShouldRetry(flags, fTimeout, timeout, elapsed)) {233 repeat = true;234 break;235 } else {236 return static_cast<int>(TransferCode::timeout);237 }238 } else {239 return zmq::HandleErrors(fId);240 }241 242 size_t moreSize = sizeof(more);243 zmq_getsockopt(fSocket, ZMQ_RCVMORE, &more, &moreSize);244 } while (more);245 246 if (repeat) {247 continue;248 }249 250 // store statistics on how many messages have been received (handle all parts as a single message)251 ++fMessagesRx;252 fBytesRx += totalSize;253 return totalSize;254 }255 }256 257 void* GetSocket() const { return fSocket; }258 259 void Close() override260 {261 // LOG(debug) << "Closing socket " << fId;262 263 closeMonitorSocket(fSocket, fMonitorSocket);264 fMonitorSocket = nullptr;265 266 if (fSocket && zmq_close(fSocket) != 0) {267 LOG(error) << "Failed closing data socket " << fId268 << ", reason: " << zmq_strerror(errno);269 }270 fSocket = nullptr;271 }272 273 void SetOption(const std::string& option, const void* value, size_t valueSize) override274 {275 if (zmq_setsockopt(fSocket, getConstant(option), value, valueSize) < 0) {276 LOG(error) << "Failed setting socket option, reason: " << zmq_strerror(errno);277 }278 }279 280 void GetOption(const std::string& option, void* value, size_t* valueSize) override281 {282 if (zmq_getsockopt(fSocket, getConstant(option), value, valueSize) < 0) {283 LOG(error) << "Failed getting socket option, reason: " << zmq_strerror(errno);284 }285 }286 287 int Events(uint32_t* events) override288 {289 size_t eventsSize = sizeof(uint32_t);290 return zmq_getsockopt(fSocket, ZMQ_EVENTS, events, &eventsSize);291 }292 293 void SetLinger(int value) override294 {295 if (zmq_setsockopt(fSocket, ZMQ_LINGER, &value, sizeof(value)) < 0) {296 throw SocketError(tools::ToString("failed setting ZMQ_LINGER, reason: ", zmq_strerror(errno)));297 }298 }299 300 int GetLinger() const override301 {302 int value = 0;303 size_t valueSize = sizeof(value);304 if (zmq_getsockopt(fSocket, ZMQ_LINGER, &value, &valueSize) < 0) {305 throw SocketError(tools::ToString("failed getting ZMQ_LINGER, reason: ", zmq_strerror(errno)));306 }307 return value;308 }309 310 void SetSndBufSize(int value) override311 {312 if (zmq_setsockopt(fSocket, ZMQ_SNDHWM, &value, sizeof(value)) < 0) {313 throw SocketError(tools::ToString("failed setting ZMQ_SNDHWM, reason: ", zmq_strerror(errno)));314 }315 }316 317 int GetSndBufSize() const override318 {319 int value = 0;320 size_t valueSize = sizeof(value);321 if (zmq_getsockopt(fSocket, ZMQ_SNDHWM, &value, &valueSize) < 0) {322 throw SocketError(tools::ToString("failed getting ZMQ_SNDHWM, reason: ", zmq_strerror(errno)));323 }324 return value;325 }326 327 void SetRcvBufSize(int value) override328 {329 if (zmq_setsockopt(fSocket, ZMQ_RCVHWM, &value, sizeof(value)) < 0) {330 throw SocketError(tools::ToString("failed setting ZMQ_RCVHWM, reason: ", zmq_strerror(errno)));331 }332 }333 334 int GetRcvBufSize() const override335 {336 int value = 0;337 size_t valueSize = sizeof(value);338 if (zmq_getsockopt(fSocket, ZMQ_RCVHWM, &value, &valueSize) < 0) {339 throw SocketError(tools::ToString("failed getting ZMQ_RCVHWM, reason: ", zmq_strerror(errno)));340 }341 return value;342 }343 344 void SetSndKernelSize(int value) override345 {346 if (zmq_setsockopt(fSocket, ZMQ_SNDBUF, &value, sizeof(value)) < 0) {347 throw SocketError(tools::ToString("failed getting ZMQ_SNDBUF, reason: ", zmq_strerror(errno)));348 }349 }350 351 int GetSndKernelSize() const override352 {353 int value = 0;354 size_t valueSize = sizeof(value);355 if (zmq_getsockopt(fSocket, ZMQ_SNDBUF, &value, &valueSize) < 0) {356 throw SocketError(tools::ToString("failed getting ZMQ_SNDBUF, reason: ", zmq_strerror(errno)));357 }358 return value;359 }360 361 void SetRcvKernelSize(int value) override362 {363 if (zmq_setsockopt(fSocket, ZMQ_RCVBUF, &value, sizeof(value)) < 0) {364 throw SocketError(tools::ToString("failed getting ZMQ_RCVBUF, reason: ", zmq_strerror(errno)));365 }366 }367 368 int GetRcvKernelSize() const override369 {370 int value = 0;371 size_t valueSize = sizeof(value);372 if (zmq_getsockopt(fSocket, ZMQ_RCVBUF, &value, &valueSize) < 0) {373 throw SocketError(tools::ToString("failed getting ZMQ_RCVBUF, reason: ", zmq_strerror(errno)));374 }375 return value;376 }377 378 unsigned long GetNumberOfConnectedPeers() const override379 {380 fConnectedPeersCount = updateNumberOfConnectedPeers(fConnectedPeersCount, fMonitorSocket);381 return fConnectedPeersCount;382 }383 384 unsigned long GetBytesTx() const override { return fBytesTx; }385 unsigned long GetBytesRx() const override { return fBytesRx; }386 unsigned long GetMessagesTx() const override { return fMessagesTx; }387 unsigned long GetMessagesRx() const override { return fMessagesRx; }388 389 [[deprecated("Use fair::mq::zmq::getConstant() from <fairmq/zeromq/Common.h> instead.")]]390 static int GetConstant(const std::string& constant) { return getConstant(constant); }391 392 ~Socket() override { Close(); }393 394 private:395 Context& fCtx;396 std::string fId;397 void* fSocket;398 void* fMonitorSocket;399 std::atomic<unsigned long> fBytesTx;400 std::atomic<unsigned long> fBytesRx;401 std::atomic<unsigned long> fMessagesTx;402 std::atomic<unsigned long> fMessagesRx;403 404 int fTimeout{100};405 mutable unsigned long fConnectedPeersCount{0};406};407 408} // namespace fair::mq::zmq409 410#endif /* FAIR_MQ_ZMQ_SOCKET_H */