Source · fairmq/zeromq/
Message.hheader
1/********************************************************************************2 * Copyright (C) 2014 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_MESSAGE_H10#define FAIR_MQ_ZMQ_MESSAGE_H11 12#include <fairmq/zeromq/UnmanagedRegion.h>13#include <fairmq/Message.h>14#include <fairmq/UnmanagedRegion.h>15 16#include <fairlogger/Logger.h>17 18#include <zmq.h>19 20#include <cstddef>21#include <cstdlib> // malloc22#include <cstring>23#include <memory> // make_unique24#include <new> // bad_alloc25#include <string>26 27namespace fair::mq::zmq28{29 30class Socket;31 32class Message final : public fair::mq::Message33{34 friend class Socket;35 36 public:37 Message(const Message&) = delete;38 Message(Message&&) = delete;39 Message& operator=(const Message&) = delete;40 Message& operator=(Message&&) = delete;41 42 Message(fair::mq::TransportFactory* factory = nullptr)43 : fair::mq::Message(factory)44 , fMsg(std::make_unique<zmq_msg_t>())45 {46 if (zmq_msg_init(fMsg.get()) != 0) {47 LOG(error) << "failed initializing message, reason: " << zmq_strerror(errno);48 }49 }50 51 Message(Alignment alignment, fair::mq::TransportFactory* factory = nullptr)52 : fair::mq::Message(factory)53 , fAlignment(alignment.alignment)54 , fMsg(std::make_unique<zmq_msg_t>())55 {56 if (zmq_msg_init(fMsg.get()) != 0) {57 LOG(error) << "failed initializing message, reason: " << zmq_strerror(errno);58 }59 }60 61 Message(const size_t size, fair::mq::TransportFactory* factory = nullptr)62 : fair::mq::Message(factory)63 , fMsg(std::make_unique<zmq_msg_t>())64 {65 if (zmq_msg_init_size(fMsg.get(), size) != 0) {66 LOG(error) << "failed initializing message with size, reason: " << zmq_strerror(errno);67 }68 }69 70 static std::pair<void*, void*> AllocateAligned(size_t size, size_t alignment)71 {72 char* fullBufferPtr = static_cast<char*>(malloc(size + alignment));73 if (!fullBufferPtr) {74 LOG(error) << "failed to allocate buffer with provided size (" << size << ") and alignment (" << alignment << ").";75 throw std::bad_alloc();76 }77 78 size_t offset = alignment - (reinterpret_cast<uintptr_t>(fullBufferPtr) % alignment);79 char* alignedPartPtr = fullBufferPtr + offset;80 81 return {static_cast<void*>(fullBufferPtr), static_cast<void*>(alignedPartPtr)};82 }83 84 Message(const size_t size, Alignment alignment, fair::mq::TransportFactory* factory = nullptr)85 : fair::mq::Message(factory)86 , fAlignment(alignment.alignment)87 , fMsg(std::make_unique<zmq_msg_t>())88 {89 if (fAlignment != 0) {90 auto ptrs = AllocateAligned(size, fAlignment);91 if (zmq_msg_init_data(fMsg.get(), ptrs.second, size, [](void* /* data */, void* hint) { free(hint); }, ptrs.first) != 0) {92 LOG(error) << "failed initializing message with size, reason: " << zmq_strerror(errno);93 }94 } else {95 if (zmq_msg_init_size(fMsg.get(), size) != 0) {96 LOG(error) << "failed initializing message with size, reason: " << zmq_strerror(errno);97 }98 }99 }100 101 Message(void* data, const size_t size, fair::mq::FreeFn* ffn, void* hint = nullptr, fair::mq::TransportFactory* factory = nullptr)102 : fair::mq::Message(factory)103 , fMsg(std::make_unique<zmq_msg_t>())104 {105 if (zmq_msg_init_data(fMsg.get(), data, size, ffn, hint) != 0) {106 LOG(error) << "failed initializing message with data, reason: " << zmq_strerror(errno);107 }108 }109 110 Message(UnmanagedRegionPtr& region,111 void* data,112 const size_t size,113 void* hint = nullptr,114 fair::mq::TransportFactory* factory = nullptr)115 : fair::mq::Message(factory)116 , fMsg(std::make_unique<zmq_msg_t>())117 {118 if (region->GetType() != GetType()) {119 LOG(error) << "region type (" << region->GetType() << ") does not match message type (" << GetType() << ")";120 throw TransportError(tools::ToString("region type (", region->GetType(), ") does not match message type (", GetType(), ")"));121 }122 123 // FIXME: make this zero-copy:124 // simply taking over the provided buffer can casue premature delete, since region could be125 // destroyed before the message is sent out. Needs lifetime extension for the ZMQ region.126 if (zmq_msg_init_size(fMsg.get(), size) != 0) {127 LOG(error) << "failed initializing message with size, reason: " << zmq_strerror(errno);128 }129 130 std::memcpy(zmq_msg_data(fMsg.get()), data, size);131 // call region callback132 auto ptr = static_cast<UnmanagedRegion*>(region.get());133 if (ptr->fBulkCallback) {134 ptr->fBulkCallback({{data, size, hint}});135 } else if (ptr->fCallback) {136 ptr->fCallback(data, size, hint);137 }138 139 // if (zmq_msg_init_data(fMsg.get(), data, size, [](void*, void*){}, nullptr) != 0)140 // {141 // LOG(error) << "failed initializing message with data, reason: " <<142 // zmq_strerror(errno);143 // }144 }145 146 void Rebuild() override147 {148 CloseMessage();149 fMsg = std::make_unique<zmq_msg_t>();150 if (zmq_msg_init(fMsg.get()) != 0) {151 LOG(error) << "failed initializing message, reason: " << zmq_strerror(errno);152 }153 }154 155 void Rebuild(Alignment alignment) override156 {157 CloseMessage();158 fAlignment = alignment.alignment;159 fMsg = std::make_unique<zmq_msg_t>();160 if (zmq_msg_init(fMsg.get()) != 0) {161 LOG(error) << "failed initializing message, reason: " << zmq_strerror(errno);162 }163 }164 165 void Rebuild(size_t size) override166 {167 CloseMessage();168 fMsg = std::make_unique<zmq_msg_t>();169 if (zmq_msg_init_size(fMsg.get(), size) != 0) {170 LOG(error) << "failed initializing message with size, reason: " << zmq_strerror(errno);171 }172 }173 174 void Rebuild(size_t size, Alignment alignment) override175 {176 CloseMessage();177 fAlignment = alignment.alignment;178 fMsg = std::make_unique<zmq_msg_t>();179 180 if (fAlignment != 0) {181 auto ptrs = AllocateAligned(size, fAlignment);182 if (zmq_msg_init_data(fMsg.get(), ptrs.second, size, [](void* /* data */, void* hint) { free(hint); }, ptrs.first) != 0) {183 LOG(error) << "failed initializing message with size, reason: " << zmq_strerror(errno);184 }185 } else {186 if (zmq_msg_init_size(fMsg.get(), size) != 0) {187 LOG(error) << "failed initializing message with size, reason: " << zmq_strerror(errno);188 }189 }190 }191 192 void Rebuild(void* data, size_t size, fair::mq::FreeFn* ffn, void* hint = nullptr) override193 {194 CloseMessage();195 fMsg = std::make_unique<zmq_msg_t>();196 if (zmq_msg_init_data(fMsg.get(), data, size, ffn, hint) != 0) {197 LOG(error) << "failed initializing message with data, reason: " << zmq_strerror(errno);198 }199 }200 201 void* GetData() const override202 {203 if (zmq_msg_size(fMsg.get()) > 0) {204 return zmq_msg_data(fMsg.get());205 } else {206 return nullptr;207 }208 }209 210 size_t GetSize() const override { return zmq_msg_size(fMsg.get()); }211 212 // To emulate shrinking, a new message is created with the new size (ViewMsg), that points to213 // the original buffer with the new size. Once the "view message" is transfered, the original is214 // destroyed. Used size is applied only once in ApplyUsedSize, which is called by the socket215 // before sending. This function just updates the desired size until the actual "resizing"216 // happens.217 bool SetUsedSize(size_t size, Alignment /* alignment */ = Alignment{0}) override218 {219 if (size == GetSize()) {220 // nothing to do221 return true;222 } else if (size > GetSize()) {223 LOG(error) << "cannot set used size higher than original.";224 return false;225 } else {226 auto newMsg = std::make_unique<zmq_msg_t>();227 void* data = GetData();228 if (zmq_msg_init_data(newMsg.get(), data, size, [](void* /* data */, void* obj) {229 zmq_msg_close(static_cast<zmq_msg_t*>(obj));230 delete static_cast<zmq_msg_t*>(obj);231 }, fMsg.get()) != 0) {232 LOG(error) << "failed initializing message with data, reason: " << zmq_strerror(errno);233 return false;234 }235 fMsg.release();236 fMsg.swap(newMsg);237 return true;238 }239 }240 241 void Realign()242 {243 // if alignment is provided244 if (fAlignment != 0) {245 void* data = GetData();246 size_t size = GetSize();247 // if buffer is valid && not already aligned with the given alignment248 if (data != nullptr && reinterpret_cast<uintptr_t>(GetData()) % fAlignment) {249 // create new aligned buffer250 auto ptrs = AllocateAligned(size, fAlignment);251 std::memcpy(ptrs.second, zmq_msg_data(fMsg.get()), size);252 // rebuild the message with the new buffer253 Rebuild(ptrs.second, size, [](void* /* buf */, void* hint) { free(hint); }, ptrs.first);254 }255 }256 }257 258 Transport GetType() const override { return Transport::ZMQ; }259 260 void Copy(const fair::mq::Message& msg) override261 {262 const Message& zMsg = static_cast<const Message&>(msg);263 // Shares the message buffer between msg and this fMsg.264 if (zmq_msg_copy(fMsg.get(), zMsg.GetMessage()) != 0) {265 LOG(error) << "failed copying message, reason: " << zmq_strerror(errno);266 return;267 }268 }269 270 ~Message() override { CloseMessage(); }271 272 private:273 size_t fAlignment = 0;274 std::unique_ptr<zmq_msg_t> fMsg;275 276 zmq_msg_t* GetMessage() const { return fMsg.get(); }277 278 void CloseMessage()279 {280 if (zmq_msg_close(fMsg.get()) != 0) {281 LOG(error) << "failed closing message, reason: " << zmq_strerror(errno);282 }283 // reset the message object to allow reuse in Rebuild284 fMsg.reset(nullptr);285 fAlignment = 0;286 }287};288 289} // namespace fair::mq::zmq290 291#endif /* FAIR_MQ_ZMQ_MESSAGE_H */