Source · fairmq/shmem/
Socket.hpublic header
Defines class fair::mq::shmem::Socket.
Described in Channels and sockets, How shared memory works, Tune and debug.
1/********************************************************************************2 * Copyright (C) 2014-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#ifndef FAIR_MQ_SHMEM_SOCKET_H_9#define FAIR_MQ_SHMEM_SOCKET_H_10 11#include "Common.h"12#include "Manager.h"13#include "Message.h"14#include <fairmq/Error.h> // for assertm15#include <fairmq/Message.h>16#include <fairmq/Socket.h>17#include <fairmq/tools/Strings.h>18#include <fairmq/zeromq/Common.h> // for zmq::HandleErrors, zmq::ShouldRetry19#include <fairmq/zeromq/ZMsg.h> // for zmq::ZMsg20 21#include <fairlogger/Logger.h>22 23#include <zmq.h>24 25#include <algorithm> // for std::max26#include <atomic>27#include <cstddef> // for std::size_t28#include <cstring> // for std::memcpy29#include <exception> // for std::terminate30#include <memory> // for std::make_unique31 32namespace fair::mq {33 class TransportFactory;34}35 36namespace fair::mq::shmem37{38 39class Socket final : public fair::mq::Socket40{41 public:42 Socket(Manager& manager,43 const std::string& type,44 const std::string& name,45 const std::string& id,46 void* context,47 fair::mq::TransportFactory* fac = nullptr)48 : fair::mq::Socket(fac)49 , fManager(manager)50 , fId(id + "." + name + "." + type)51 , fBytesTx(0)52 , fBytesRx(0)53 , fMessagesTx(0)54 , fMessagesRx(0)55 , fMetadataMsgSize(manager.GetMetadataMsgSize())56 {57 assert(context);58 59 if (type == "sub" || type == "pub") {60 LOG(error) << "PUB/SUB socket type is not supported for shared memory transport";61 throw SocketError("PUB/SUB socket type is not supported for shared memory transport");62 }63 64 fSocket = zmq_socket(context, zmq::getConstant(type));65 fMonitorSocket = zmq::makeMonitorSocket(context, fSocket, fId);66 67 if (fSocket == nullptr) {68 LOG(error) << "Failed creating socket " << fId << ", reason: " << zmq_strerror(errno);69 throw SocketError(tools::ToString("Failed creating socket ", fId, ", reason: ", zmq_strerror(errno)));70 }71 72 if (zmq_setsockopt(fSocket, ZMQ_IDENTITY, fId.c_str(), fId.length()) != 0) {73 LOG(error) << "Failed setting ZMQ_IDENTITY socket option, reason: " << zmq_strerror(errno);74 }75 76 // Tell socket to try and send/receive outstanding messages for <linger> milliseconds before terminating.77 // Default value for ZeroMQ is -1, which is to wait forever.78 int linger = 1000;79 if (zmq_setsockopt(fSocket, ZMQ_LINGER, &linger, sizeof(linger)) != 0) {80 LOG(error) << "Failed setting ZMQ_LINGER socket option, reason: " << zmq_strerror(errno);81 }82 83 if (zmq_setsockopt(fSocket, ZMQ_SNDTIMEO, &fTimeout, sizeof(fTimeout)) != 0) {84 LOG(error) << "Failed setting ZMQ_SNDTIMEO socket option, reason: " << zmq_strerror(errno);85 }86 87 if (zmq_setsockopt(fSocket, ZMQ_RCVTIMEO, &fTimeout, sizeof(fTimeout)) != 0) {88 LOG(error) << "Failed setting ZMQ_RCVTIMEO socket option, reason: " << zmq_strerror(errno);89 }90 91 // if (type == "sub")92 // {93 // if (zmq_setsockopt(fSocket, ZMQ_SUBSCRIBE, nullptr, 0) != 0)94 // {95 // LOG(error) << "Failed setting ZMQ_SUBSCRIBE socket option, reason: " << zmq_strerror(errno);96 // }97 // }98 LOG(debug) << "Created socket " << GetId();99 }100 101 Socket(const Socket&) = delete;102 Socket(Socket&&) = delete;103 Socket& operator=(const Socket&) = delete;104 Socket& operator=(Socket&&) = delete;105 106 std::string GetId() const override { return fId; }107 108 bool Bind(const std::string& address) override109 {110 return zmq::Bind(fSocket, address, fId);111 }112 113 bool Connect(const std::string& address) override114 {115 return zmq::Connect(fSocket, address, fId);116 }117 118 int64_t Send(mq::MessagePtr& msg, int timeout = -1) override119 {120 auto msgPtr = msg.get();121 if (!msgPtr) {122 return static_cast<int>(TransferCode::error);123 }124 assertm(dynamic_cast<shmem::Message*>(msgPtr), "given mq::Message is a shmem::Message"); // NOLINT125 auto shmMsg = static_cast<shmem::Message*>(msgPtr); // NOLINT(cppcoreguidelines-pro-type-static-cast-downcast)126 127 int flags = 0;128 if (timeout == 0) {129 flags = ZMQ_DONTWAIT;130 }131 int elapsed = 0;132 133 MetaHeader meta{ shmMsg->fSize, shmMsg->fHint, shmMsg->fHandle, shmMsg->fShared, shmMsg->fRegionId, shmMsg->fSegmentId, shmMsg->fManaged };134 135 // meta msg format: | MetaHeader | padded to fMetadataMsgSize |136 zmq::ZMsg zmqMsg(std::max(fMetadataMsgSize, sizeof(MetaHeader)));137 std::memcpy(zmqMsg.Data(), &meta, sizeof(MetaHeader));138 139 while (true) {140 int nbytes = zmq_msg_send(zmqMsg.Msg(), fSocket, flags);141 if (nbytes > 0) {142 shmMsg->fQueued = true;143 ++fMessagesTx;144 size_t size = msg->GetSize();145 fBytesTx += size;146 return size;147 } else if (zmq_errno() == EAGAIN || zmq_errno() == EINTR) {148 if (fManager.Interrupted()) {149 return static_cast<int>(TransferCode::interrupted);150 } else if (zmq::ShouldRetry(flags, fTimeout, timeout, elapsed)) {151 continue;152 } else {153 return static_cast<int>(TransferCode::timeout);154 }155 } else {156 return zmq::HandleErrors(fId);157 }158 }159 160 return static_cast<int>(TransferCode::error);161 }162 163 int64_t Receive(MessagePtr& msg, int timeout = -1) override164 {165 int flags = 0;166 if (timeout == 0) {167 flags = ZMQ_DONTWAIT;168 }169 int elapsed = 0;170 171 while (true) {172 Message* shmMsg = static_cast<Message*>(msg.get());173 MetaHeader meta;174 int nbytes = zmq_recv(fSocket, &meta, sizeof(MetaHeader), flags);175 if (nbytes > 0) {176 // check for number of received messages. must be 1177 if (static_cast<std::size_t>(nbytes) < sizeof(MetaHeader)) {178 throw SocketError(179 tools::ToString("Received message is not a valid FairMQ shared memory message. ",180 "Possibly due to a misconfigured transport on the sender side. ",181 "Expected minimum size of ", sizeof(MetaHeader), " bytes, received ", nbytes));182 }183 184 shmMsg->SetMeta(meta);185 186 size_t size = shmMsg->GetSize();187 fBytesRx += size;188 ++fMessagesRx;189 return size;190 } else if (zmq_errno() == EAGAIN || zmq_errno() == EINTR) {191 if (fManager.Interrupted()) {192 return static_cast<int>(TransferCode::interrupted);193 } else if (zmq::ShouldRetry(flags, fTimeout, timeout, elapsed)) {194 continue;195 } else {196 return static_cast<int>(TransferCode::timeout);197 }198 } else {199 return zmq::HandleErrors(fId);200 }201 }202 }203 204 int64_t Send(Parts::container& msgVec, int timeout = -1) override205 {206 int flags = 0;207 if (timeout == 0) {208 flags = ZMQ_DONTWAIT;209 }210 int elapsed = 0;211 212 // meta msg format: | n | MetaHeader 1 | ... | MetaHeader n | padded to fMetadataMsgSize |213 auto const n = msgVec.size();214 zmq::ZMsg zmqMsg(std::max(fMetadataMsgSize, sizeof(std::size_t) + n * sizeof(MetaHeader)));215 216 auto meta_n = static_cast<std::size_t*>(zmqMsg.Data());217 *meta_n = n;218 ++meta_n;219 auto metas = static_cast<MetaHeader*>(static_cast<void*>(meta_n));220 for (auto& msg : msgVec) {221 auto msgPtr = msg.get();222 if (!msgPtr) {223 return static_cast<int>(TransferCode::error);224 }225 assertm(dynamic_cast<shmem::Message*>(msgPtr), "given mq::Message is a shmem::Message"); // NOLINT226 auto shmMsg = static_cast<shmem::Message*>(msgPtr); // NOLINT(cppcoreguidelines-pro-type-static-cast-downcast)227 MetaHeader meta{ shmMsg->fSize, shmMsg->fHint, shmMsg->fHandle, shmMsg->fShared, shmMsg->fRegionId, shmMsg->fSegmentId, shmMsg->fManaged };228 std::memcpy(metas++, &meta, sizeof(MetaHeader));229 }230 231 while (true) {232 int64_t totalSize = 0;233 int nbytes = zmq_msg_send(zmqMsg.Msg(), fSocket, flags);234 if (nbytes > 0) {235 assert(static_cast<unsigned int>(nbytes) >= sizeof(std::size_t) + (n * sizeof(MetaHeader)));236 237 for (auto& msg : msgVec) {238 Message* shmMsg = static_cast<Message*>(msg.get());239 shmMsg->fQueued = true;240 totalSize += shmMsg->fSize;241 }242 243 // store statistics on how many messages have been sent244 fMessagesTx++;245 fBytesTx += totalSize;246 247 return totalSize;248 } else if (zmq_errno() == EAGAIN || zmq_errno() == EINTR) {249 if (fManager.Interrupted()) {250 return static_cast<int>(TransferCode::interrupted);251 } else if (zmq::ShouldRetry(flags, fTimeout, timeout, elapsed)) {252 continue;253 } else {254 return static_cast<int>(TransferCode::timeout);255 }256 } else {257 return zmq::HandleErrors(fId);258 }259 }260 261 return static_cast<int>(TransferCode::error);262 }263 264 int64_t Receive(Parts::container& msgVec, int timeout = -1) override265 {266 int flags = 0;267 if (timeout == 0) {268 flags = ZMQ_DONTWAIT;269 }270 int elapsed = 0;271 272 zmq::ZMsg zmqMsg;273 274 while (true) {275 std::size_t totalSize = 0;276 int nbytes = zmq_msg_recv(zmqMsg.Msg(), fSocket, flags);277 if (nbytes > 0) {278 [[maybe_unused]] auto const size = zmqMsg.Size();279 assert(size > sizeof(std::size_t));280 auto meta_n = static_cast<std::size_t*>(zmqMsg.Data());281 auto const n = *meta_n;282 assert(size >= sizeof(std::size_t) + n * sizeof(MetaHeader));283 ++meta_n;284 auto metas = static_cast<MetaHeader*>(static_cast<void*>(meta_n));285 msgVec.reserve(msgVec.size() + n);286 auto const transport = GetTransport();287 288 for (std::size_t i = 0; i < n; ++i) {289 msgVec.push_back(std::make_unique<Message>(fManager, *metas, transport));290 ++metas;291 // NOLINTNEXTLINE(cppcoreguidelines-pro-type-static-cast-downcast)292 Message* shmMsg = static_cast<Message*>(msgVec.back().get());293 totalSize += shmMsg->GetSize();294 }295 296 // store statistics on how many messages have been received (handle all parts as a single message)297 fMessagesRx++;298 fBytesRx += totalSize;299 300 return totalSize;301 } else if (zmq_errno() == EAGAIN || zmq_errno() == EINTR) {302 if (fManager.Interrupted()) {303 return static_cast<int>(TransferCode::interrupted);304 } else if (zmq::ShouldRetry(flags, fTimeout, timeout, elapsed)) {305 continue;306 } else {307 return static_cast<int>(TransferCode::timeout);308 }309 } else {310 return zmq::HandleErrors(fId);311 }312 }313 314 return static_cast<int>(TransferCode::error);315 }316 317 void* GetSocket() const { return fSocket; }318 319 void Close() override320 {321 // LOG(debug) << "Closing socket " << fId;322 323 zmq::closeMonitorSocket(fSocket, fMonitorSocket);324 fMonitorSocket = nullptr;325 326 if (fSocket && zmq_close(fSocket) != 0) {327 LOG(error) << "Failed closing data socket " << fId328 << ", reason: " << zmq_strerror(errno);329 }330 fSocket = nullptr;331 }332 333 void SetOption(const std::string& option, const void* value, size_t valueSize) override334 {335 if (zmq_setsockopt(fSocket, zmq::getConstant(option), value, valueSize) < 0) {336 LOG(error) << "Failed setting socket option, reason: " << zmq_strerror(errno);337 }338 }339 340 void GetOption(const std::string& option, void* value, size_t* valueSize) override341 {342 if (zmq_getsockopt(fSocket, zmq::getConstant(option), value, valueSize) < 0) {343 LOG(error) << "Failed getting socket option, reason: " << zmq_strerror(errno);344 }345 }346 347 void SetLinger(int value) override348 {349 if (zmq_setsockopt(fSocket, ZMQ_LINGER, &value, sizeof(value)) < 0) {350 throw SocketError(tools::ToString("failed setting ZMQ_LINGER, reason: ", zmq_strerror(errno)));351 }352 }353 354 int Events(uint32_t* events) override355 {356 size_t eventsSize = sizeof(uint32_t);357 return zmq_getsockopt(fSocket, ZMQ_EVENTS, events, &eventsSize);358 }359 360 int GetLinger() const override361 {362 int value = 0;363 size_t valueSize = sizeof(value);364 if (zmq_getsockopt(fSocket, ZMQ_LINGER, &value, &valueSize) < 0) {365 throw SocketError(tools::ToString("failed getting ZMQ_LINGER, reason: ", zmq_strerror(errno)));366 }367 return value;368 }369 370 void SetSndBufSize(int value) override371 {372 if (zmq_setsockopt(fSocket, ZMQ_SNDHWM, &value, sizeof(value)) < 0) {373 throw SocketError(tools::ToString("failed setting ZMQ_SNDHWM, reason: ", zmq_strerror(errno)));374 }375 }376 377 int GetSndBufSize() const override378 {379 int value = 0;380 size_t valueSize = sizeof(value);381 if (zmq_getsockopt(fSocket, ZMQ_SNDHWM, &value, &valueSize) < 0) {382 throw SocketError(tools::ToString("failed getting ZMQ_SNDHWM, reason: ", zmq_strerror(errno)));383 }384 return value;385 }386 387 void SetRcvBufSize(int value) override388 {389 if (zmq_setsockopt(fSocket, ZMQ_RCVHWM, &value, sizeof(value)) < 0) {390 throw SocketError(tools::ToString("failed setting ZMQ_RCVHWM, reason: ", zmq_strerror(errno)));391 }392 }393 394 int GetRcvBufSize() const override395 {396 int value = 0;397 size_t valueSize = sizeof(value);398 if (zmq_getsockopt(fSocket, ZMQ_RCVHWM, &value, &valueSize) < 0) {399 throw SocketError(tools::ToString("failed getting ZMQ_RCVHWM, reason: ", zmq_strerror(errno)));400 }401 return value;402 }403 404 void SetSndKernelSize(int value) override405 {406 if (zmq_setsockopt(fSocket, ZMQ_SNDBUF, &value, sizeof(value)) < 0) {407 throw SocketError(tools::ToString("failed getting ZMQ_SNDBUF, reason: ", zmq_strerror(errno)));408 }409 }410 411 int GetSndKernelSize() const override412 {413 int value = 0;414 size_t valueSize = sizeof(value);415 if (zmq_getsockopt(fSocket, ZMQ_SNDBUF, &value, &valueSize) < 0) {416 throw SocketError(tools::ToString("failed getting ZMQ_SNDBUF, reason: ", zmq_strerror(errno)));417 }418 return value;419 }420 421 void SetRcvKernelSize(int value) override422 {423 if (zmq_setsockopt(fSocket, ZMQ_RCVBUF, &value, sizeof(value)) < 0) {424 throw SocketError(tools::ToString("failed getting ZMQ_RCVBUF, reason: ", zmq_strerror(errno)));425 }426 }427 428 int GetRcvKernelSize() const override429 {430 int value = 0;431 size_t valueSize = sizeof(value);432 if (zmq_getsockopt(fSocket, ZMQ_RCVBUF, &value, &valueSize) < 0) {433 throw SocketError(tools::ToString("failed getting ZMQ_RCVBUF, reason: ", zmq_strerror(errno)));434 }435 return value;436 }437 438 unsigned long GetNumberOfConnectedPeers() const override439 {440 fConnectedPeersCount = zmq::updateNumberOfConnectedPeers(fConnectedPeersCount, fMonitorSocket);441 return fConnectedPeersCount;442 }443 444 unsigned long GetBytesTx() const override { return fBytesTx; }445 unsigned long GetBytesRx() const override { return fBytesRx; }446 unsigned long GetMessagesTx() const override { return fMessagesTx; }447 unsigned long GetMessagesRx() const override { return fMessagesRx; }448 449 [[deprecated("Use fair::mq::zmq::getConstant() from <fairmq/zeromq/Common.h> instead.")]]450 static int GetConstant(const std::string& constant) { return zmq::getConstant(constant); }451 452 ~Socket() override { Close(); }453 454 private:455 Manager& fManager;456 std::string fId;457 void* fSocket{nullptr};458 void* fMonitorSocket{nullptr};459 std::atomic<unsigned long> fBytesTx;460 std::atomic<unsigned long> fBytesRx;461 std::atomic<unsigned long> fMessagesTx;462 std::atomic<unsigned long> fMessagesRx;463 464 int fTimeout{100};465 mutable unsigned long fConnectedPeersCount{0};466 std::size_t fMetadataMsgSize;467};468 469} // namespace fair::mq::shmem470 471#endif /* FAIR_MQ_SHMEM_SOCKET_H_ */