Source · fairmq/shmem/
Message.hpublic header
Defines class fair::mq::shmem::Message.
Described in readout: a readout chain, region: unmanaged regions, Handle errors and timeouts, How shared memory works, Unmanaged regions, Out-of-band buffers.
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#ifndef FAIR_MQ_SHMEM_MESSAGE_H_9#define FAIR_MQ_SHMEM_MESSAGE_H_10 11#include "Common.h"12#include "Manager.h"13#include "UnmanagedRegion.h"14#include "UnmanagedRegionImpl.h"15#include <fairmq/Message.h>16#include <fairmq/UnmanagedRegion.h>17#include <fairmq/Transports.h>18 19#include <fairlogger/Logger.h>20 21#include <boost/interprocess/mapped_region.hpp>22 23#include <cstddef> // size_t24#include <atomic>25 26#include <sys/types.h> // getpid27#include <unistd.h> // pid_t28 29namespace fair::mq::shmem30{31 32class Socket;33 34class Message final : public fair::mq::Message35{36 friend class Socket;37 38 public:39 Message(Manager& manager, fair::mq::TransportFactory* factory = nullptr)40 : Message(manager, Alignment{0}, factory)41 {}42 43 Message(Manager& manager, Alignment /* alignment */, fair::mq::TransportFactory* factory = nullptr)44 : fair::mq::Message(factory)45 , fManager(manager)46 , fSegmentId(fManager.GetSegmentId())47 {48 fManager.IncrementMsgCounter();49 }50 51 Message(Manager& manager, const size_t size, fair::mq::TransportFactory* factory = nullptr)52 : Message(manager, size, Alignment{0}, factory)53 {}54 55 Message(Manager& manager, const size_t size, Alignment alignment, fair::mq::TransportFactory* factory = nullptr)56 : fair::mq::Message(factory)57 , fManager(manager)58 , fSegmentId(fManager.GetSegmentId())59 {60 InitializeChunk(size, alignment.alignment);61 fManager.IncrementMsgCounter();62 }63 64 Message(Manager& manager, void* data, const size_t size, fair::mq::FreeFn* ffn, void* hint = nullptr, fair::mq::TransportFactory* factory = nullptr)65 : fair::mq::Message(factory)66 , fManager(manager)67 , fSegmentId(fManager.GetSegmentId())68 {69 if (InitializeChunk(size)) {70 std::memcpy(fLocalPtr, data, size);71 if (ffn) {72 ffn(data, hint);73 } else {74 free(data);75 }76 }77 fManager.IncrementMsgCounter();78 }79 80 Message(Manager& manager,81 UnmanagedRegionPtr& region,82 void* data,83 const size_t size,84 void* hint = nullptr,85 fair::mq::TransportFactory* factory = nullptr)86 : fair::mq::Message(factory)87 , fManager(manager)88 , fLocalPtr(static_cast<char*>(data))89 , fSize(size)90 , fHint(reinterpret_cast<size_t>(hint))91 , fRegionId(static_cast<UnmanagedRegionImpl*>(region.get())->fRegionId)92 , fSegmentId(fManager.GetSegmentId())93 , fManaged(false)94 {95 if (region->GetType() != GetType()) {96 LOG(error) << "region type (" << region->GetType() << ") does not match message type (" << GetType() << ")";97 throw TransportError(tools::ToString("region type (", region->GetType(), ") does not match message type (", GetType(), ")"));98 }99 100 if (reinterpret_cast<const char*>(data) >= reinterpret_cast<const char*>(region->GetData()) &&101 reinterpret_cast<const char*>(data) <= reinterpret_cast<const char*>(region->GetData()) + region->GetSize()) {102 fHandle = (boost::interprocess::managed_shared_memory::handle_t)(reinterpret_cast<const char*>(data) - reinterpret_cast<const char*>(region->GetData()));103 } else {104 LOG(error) << "trying to create region message with data from outside the region";105 throw TransportError("trying to create region message with data from outside the region");106 }107 fManager.IncrementMsgCounter();108 }109 110 Message(Manager& manager, MetaHeader& hdr, fair::mq::TransportFactory* factory = nullptr)111 : fair::mq::Message(factory)112 , fManager(manager)113 , fSize(hdr.fSize)114 , fHint(hdr.fHint)115 , fHandle(hdr.fHandle)116 , fShared(hdr.fShared)117 , fRegionId(hdr.fRegionId)118 , fSegmentId(hdr.fSegmentId)119 , fManaged(hdr.fManaged)120 {121 fManager.IncrementMsgCounter();122 }123 124 Message(const Message&) = delete;125 Message(Message&&) = delete;126 Message& operator=(const Message&) = delete;127 Message& operator=(Message&&) = delete;128 129 void Rebuild() override130 {131 CloseMessage();132 fQueued = false;133 }134 135 void Rebuild(Alignment /* alignment */) override136 {137 CloseMessage();138 fQueued = false;139 }140 141 void Rebuild(size_t size) override142 {143 CloseMessage();144 fQueued = false;145 InitializeChunk(size);146 }147 148 void Rebuild(size_t size, Alignment alignment) override149 {150 CloseMessage();151 fQueued = false;152 InitializeChunk(size, alignment.alignment);153 }154 155 void Rebuild(void* data, size_t size, fair::mq::FreeFn* ffn, void* hint = nullptr) override156 {157 CloseMessage();158 fQueued = false;159 160 if (InitializeChunk(size)) {161 std::memcpy(fLocalPtr, data, size);162 if (ffn) {163 ffn(data, hint);164 } else {165 free(data);166 }167 }168 }169 170 MetaHeader GetMeta() const171 {172 return {fSize, fHint, fHandle, fShared, fRegionId, fSegmentId, fManaged};173 }174 175 void* GetData() const override176 {177 if (!fLocalPtr) {178 if (fManaged) {179 if (fSize > 0) {180 fManager.GetSegment(fSegmentId);181 fLocalPtr = ShmHeader::UserPtr(fManager.GetAddressFromHandle(fHandle, fSegmentId));182 } else {183 fLocalPtr = nullptr;184 }185 } else {186 UnmanagedRegion* region = fManager.GetRegionFromCache(fRegionId);187 if (region) {188 fLocalPtr = reinterpret_cast<char*>(region->GetData()) + fHandle;189 } else {190 // LOG(warn) << "could not get pointer from a region message";191 fLocalPtr = nullptr;192 }193 }194 }195 196 return static_cast<void*>(fLocalPtr);197 }198 199 size_t GetSize() const override { return fSize; }200 201 bool SetUsedSize(size_t newSize, Alignment alignment = Alignment{0}) override202 {203 if (newSize == fSize) {204 return true;205 } else if (newSize == 0) {206 Deallocate();207 return true;208 } else if (newSize <= fSize) {209 try {210 char* oldPtr = fManager.GetAddressFromHandle(fHandle, fSegmentId);211 try {212 uint16_t userOffset = ShmHeader::UserOffset(oldPtr);213 char* ptr = fManager.ShrinkInPlace(userOffset + newSize, oldPtr, fSegmentId);214 fLocalPtr = ShmHeader::UserPtr(ptr);215 fSize = newSize;216 return true;217 } catch (boost::interprocess::bad_alloc& e) {218 // if shrinking fails (can happen due to boost alignment requirements):219 // unused size >= 1000000 bytes: reallocate fully220 // unused size < 1000000 bytes: simply reset the size and keep the rest of the buffer until message destruction221 if (fSize - newSize >= 1000000) {222 if (alignment.alignment == 0) {223 // if no alignment is provided, take the minimum alignment of the old pointer, but no more than 4096224 alignment.alignment = 1 << std::min(__builtin_ctz(reinterpret_cast<size_t>(oldPtr)), 12);225 }226 char* ptr = fManager.Allocate(newSize, alignment.alignment);227 char* userPtr = ShmHeader::UserPtr(ptr);228 std::memcpy(userPtr, fLocalPtr, newSize);229 fManager.Deallocate(fHandle, fSegmentId);230 fLocalPtr = userPtr;231 fHandle = fManager.GetHandleFromAddress(ptr, fSegmentId);232 }233 fSize = newSize;234 return true;235 }236 } catch (boost::interprocess::interprocess_exception& e) {237 LOG(debug) << "could not set used size: " << e.what();238 return false;239 }240 } else {241 LOG(error) << "cannot set used size higher than original.";242 return false;243 }244 }245 246 Transport GetType() const override { return fair::mq::Transport::SHM; }247 248 uint16_t GetRefCount() const249 {250 if (fHandle < 0) {251 return 1;252 }253 254 if (fManaged) { // managed segment255 fManager.GetSegment(fSegmentId);256 return ShmHeader::RefCount(fManager.GetAddressFromHandle(fHandle, fSegmentId));257 }258 if (fShared < 0) { // UR msg is not yet shared259 return 1;260 }261 UnmanagedRegion* region = fManager.GetRegionFromCache(fRegionId);262 if (!region) {263 throw TransportError(tools::ToString("Cannot get unmanaged region with id ", fRegionId));264 }265 if (region->fRcSegmentSize > 0) {266 return region->GetRefCountAddressFromHandle(fShared)->Get();267 } else {268 fManager.GetSegment(fSegmentId);269 return ShmHeader::RefCount(fManager.GetAddressFromHandle(fShared, fSegmentId));270 }271 }272 273 void Copy(const fair::mq::Message& other) override274 {275 const Message& otherMsg = static_cast<const Message&>(other);276 // if the other message is not initialized, close this one too and return277 if (otherMsg.fHandle < 0) {278 CloseMessage();279 return;280 }281 282 // if this msg is already initialized, close it first283 if (fHandle >= 0) {284 CloseMessage();285 }286 287 // increment ref count288 if (otherMsg.fManaged) { // msg in managed segment289 fManager.GetSegment(otherMsg.fSegmentId);290 ShmHeader::IncrementRefCount(fManager.GetAddressFromHandle(otherMsg.fHandle, otherMsg.fSegmentId));291 } else { // msg in unmanaged region292 UnmanagedRegion* region = fManager.GetRegionFromCache(otherMsg.fRegionId);293 if (!region) {294 throw TransportError(tools::ToString("Cannot get unmanaged region with id ", otherMsg.fRegionId));295 }296 if (region->fRcSegmentSize > 0) {297 if (otherMsg.fShared < 0) {298 // UR msg not yet shared, create the reference counting object with count 2299 try {300 otherMsg.fShared = region->HandleFromAddress(&(region->MakeRefCount(2)));301 } catch (boost::interprocess::bad_alloc& ba) {302 throw RefCountBadAlloc(tools::ToString("Insufficient space in the reference count segment ", otherMsg.fRegionId, ", original exception: bad_alloc: ", ba.what()));303 }304 } else {305 region->GetRefCountAddressFromHandle(otherMsg.fShared)->Increment();306 }307 } else { // if RefCount segment size is 0, store the ref count in the managed segment308 if (otherMsg.fShared < 0) { // if UR msg is not yet shared309 char* ptr = fManager.Allocate(2, 0);310 // point the fShared in the unmanaged region message to the refCount holder311 otherMsg.fShared = fManager.GetHandleFromAddress(ptr, fSegmentId);312 // the message needs to be able to locate in which segment the refCount is stored313 otherMsg.fSegmentId = fSegmentId;314 ShmHeader::IncrementRefCount(ptr);315 } else { // if the UR msg is already shared316 fManager.GetSegment(otherMsg.fSegmentId);317 ShmHeader::IncrementRefCount(fManager.GetAddressFromHandle(otherMsg.fShared, otherMsg.fSegmentId));318 }319 }320 }321 322 // copy meta data323 fSize = otherMsg.fSize;324 fHint = otherMsg.fHint;325 fHandle = otherMsg.fHandle;326 fShared = otherMsg.fShared;327 fRegionId = otherMsg.fRegionId;328 fSegmentId = otherMsg.fSegmentId;329 fManaged = otherMsg.fManaged;330 }331 332 ~Message() override { CloseMessage(); }333 334 private:335 // Members are sorted by size to avoid padding. Keep the message small: it is336 // allocated for every transfer. sizeof(Message) is checked in test/message.337 Manager& fManager;338 mutable char* fLocalPtr = nullptr; // local cache of the data ptr, for received messages obtained from the handle339 size_t fSize = 0; // size of the shm buffer340 size_t fHint = 0; // user-defined value, given by the user on message creation and returned to the user on "buffer no longer needed"-callbacks341 boost::interprocess::managed_shared_memory::handle_t fHandle = -1; // handle to shm buffer, convertible to shm buffer ptr342 mutable boost::interprocess::managed_shared_memory::handle_t fShared = -1; // handle to the buffer storing the ref count for shared buffers343 uint16_t fRegionId = 0; // id of the unmanaged region344 mutable uint16_t fSegmentId; // id of the managed segment345 bool fManaged = true; // true = managed segment, false = unmanaged region346 bool fQueued = false;347 348 void SetMeta(const MetaHeader& meta)349 {350 fSize = meta.fSize;351 fHint = meta.fHint;352 fHandle = meta.fHandle;353 fShared = meta.fShared;354 fRegionId = meta.fRegionId;355 fSegmentId = meta.fSegmentId;356 fManaged = meta.fManaged;357 }358 359 char* InitializeChunk(const size_t size, size_t alignment = 0)360 {361 if (size == 0) {362 fSize = 0;363 return fLocalPtr;364 }365 char* ptr = fManager.Allocate(size, alignment);366 fHandle = fManager.GetHandleFromAddress(ptr, fSegmentId);367 fSize = size;368 fLocalPtr = ShmHeader::UserPtr(ptr);369 return fLocalPtr;370 }371 372 void Deallocate()373 {374 if (fHandle >= 0 && !fQueued) {375 if (fManaged) { // managed segment376 fManager.GetSegment(fSegmentId);377 uint16_t refCount = ShmHeader::DecrementRefCount(fManager.GetAddressFromHandle(fHandle, fSegmentId));378 if (refCount == 1) {379 fManager.Deallocate(fHandle, fSegmentId);380 }381 } else { // unmanaged region382 UnmanagedRegion* region = fManager.GetRegionFromCache(fRegionId);383 if (fShared >= 0) {384 if (!region) {385 throw TransportError(tools::ToString("Cannot get unmanaged region with id ", fRegionId));386 }387 if (region->fRcSegmentSize > 0) {388 uint16_t refCount = region->GetRefCountAddressFromHandle(fShared)->Decrement();389 if (refCount == 1) {390 region->RemoveRefCount(*(region->GetRefCountAddressFromHandle(fShared)));391 ReleaseUnmanagedRegionBlock(region);392 }393 } else { // if RefCount segment size is 0, get the ref count from the managed segment394 // make sure segment is initialized in this transport395 fManager.GetSegment(fSegmentId);396 // release unmanaged region block if ref count is one397 uint16_t refCount = ShmHeader::DecrementRefCount(fManager.GetAddressFromHandle(fShared, fSegmentId));398 if (refCount == 1) {399 fManager.Deallocate(fShared, fSegmentId);400 ReleaseUnmanagedRegionBlock(region);401 }402 }403 } else {404 ReleaseUnmanagedRegionBlock(region);405 }406 }407 }408 fHandle = -1;409 fLocalPtr = nullptr;410 fSize = 0;411 }412 413 void ReleaseUnmanagedRegionBlock(UnmanagedRegion* region)414 {415 if (region) {416 region->ReleaseBlock({fHandle, fSize, fHint});417 } else {418 LOG(warn) << "region ack queue for id " << fRegionId << " no longer exist. Not sending ack";419 }420 }421 422 void CloseMessage()423 {424 try {425 Deallocate();426 fManager.DecrementMsgCounter();427 } catch (SharedMemoryError& sme) {428 LOG(error) << "error closing message: " << sme.what();429 } catch (boost::interprocess::lock_exception& le) {430 LOG(error) << "error closing message: " << le.what();431 }432 }433};434 435} // namespace fair::mq::shmem436 437#endif /* FAIR_MQ_SHMEM_MESSAGE_H_ */