Mirror/FairMQ
1
0
Fork 0
mirror of https://github.com/FairRootGroup/FairMQ.git synced 2025-10-13 16:46:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

shm: reimplement alignment

Browse Source
This commit is contained in:
Alexey Rybalchenko 2021-07-02 13:26:58 +02:00 committed by Dennis Klein
parent 4e8f247a0d
commit 815b2f1d76
2 changed files with 53 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

49
fairmq/shmem/Manager.h
View File

@ -33,8 +33,11 @@
#include <boost/interprocess/sync/named_mutex.hpp> #include <boost/interprocess/sync/named_mutex.hpp>
#include <boost/variant.hpp> #include <boost/variant.hpp>
#include <algorithm> // max
#include <condition_variable> #include <condition_variable>
#include <cstddef> // max_align_t
#include <cstdlib> // getenv #include <cstdlib> // getenv
#include <cstring> // memcpy
#include <memory> // make_unique #include <memory> // make_unique
#include <mutex> #include <mutex>
#include <set> #include <set>
@ -55,6 +58,25 @@
namespace fair::mq::shmem namespace fair::mq::shmem
{ {
struct ShmPtr
{
explicit ShmPtr(char* rPtr)
: realPtr(rPtr)
{}
char* RealPtr()
{
return realPtr;
}
char* UserPtr()
{
return realPtr + sizeof(uint16_t) + *(reinterpret_cast<uint16_t*>(realPtr));
}
char* realPtr;
};
class Manager class Manager
{ {
public: public:
@ -618,9 +640,13 @@ class Manager
return boost::apply_visitor(SegmentAddressFromHandle(handle), fSegments.at(segmentId)); return boost::apply_visitor(SegmentAddressFromHandle(handle), fSegments.at(segmentId));
} }
char* Allocate(const size_t size, size_t alignment = 0) ShmPtr Allocate(size_t size, size_t alignment = 0)
{ {
alignment = std::max(alignment, alignof(std::max_align_t));
char* ptr = nullptr; char* ptr = nullptr;
// [offset(uint16_t)][alignment][buffer]
size_t fullSize = sizeof(uint16_t) + alignment + size;
// tools::RateLimiter rateLimiter(20); // tools::RateLimiter rateLimiter(20);
while (ptr == nullptr) { while (ptr == nullptr) {
@ -629,14 +655,15 @@ class Manager
// char* hint = 0; // unused for boost::interprocess::allocate_new // char* hint = 0; // unused for boost::interprocess::allocate_new
// ptr = fSegments.at(fSegmentId).allocation_command<char>(boost::interprocess::allocate_new, size, actualSize, hint); // ptr = fSegments.at(fSegmentId).allocation_command<char>(boost::interprocess::allocate_new, size, actualSize, hint);
size_t segmentSize = boost::apply_visitor(SegmentSize(), fSegments.at(fSegmentId)); size_t segmentSize = boost::apply_visitor(SegmentSize(), fSegments.at(fSegmentId));
if (size > segmentSize) { if (fullSize > segmentSize) {
throw MessageBadAlloc(tools::ToString("Requested message size (", size, ") exceeds segment size (", segmentSize, ")")); throw MessageBadAlloc(tools::ToString("Requested message size (", fullSize, ") exceeds segment size (", segmentSize, ")"));
}
if (alignment == 0) {
ptr = reinterpret_cast<char*>(boost::apply_visitor(SegmentAllocate{size}, fSegments.at(fSegmentId)));
} else {
ptr = reinterpret_cast<char*>(boost::apply_visitor(SegmentAllocateAligned{size, alignment}, fSegments.at(fSegmentId)));
} }
ptr = reinterpret_cast<char*>(boost::apply_visitor(SegmentAllocate{fullSize}, fSegments.at(fSegmentId)));
assert(reinterpret_cast<uintptr_t>(ptr) % 2 == 0);
uint16_t offset = 0;
offset = alignment - ((reinterpret_cast<uintptr_t>(ptr) + sizeof(uint16_t)) % alignment);
std::memcpy(ptr, &offset, sizeof(offset));
} catch (boost::interprocess::bad_alloc& ba) { } catch (boost::interprocess::bad_alloc& ba) {
// LOG(warn) << "Shared memory full..."; // LOG(warn) << "Shared memory full...";
if (ThrowingOnBadAlloc()) { if (ThrowingOnBadAlloc()) {
@ -645,7 +672,7 @@ class Manager
// rateLimiter.maybe_sleep(); // rateLimiter.maybe_sleep();
std::this_thread::sleep_for(std::chrono::milliseconds(50)); std::this_thread::sleep_for(std::chrono::milliseconds(50));
if (Interrupted()) { if (Interrupted()) {
return ptr; return ShmPtr(ptr);
} else { } else {
continue; continue;
} }
@ -657,13 +684,13 @@ class Manager
(*fMsgDebug).emplace(fSegmentId, fShmVoidAlloc); (*fMsgDebug).emplace(fSegmentId, fShmVoidAlloc);
} }
(*fMsgDebug).at(fSegmentId).emplace( (*fMsgDebug).at(fSegmentId).emplace(
static_cast<size_t>(GetHandleFromAddress(ptr, fSegmentId)), static_cast<size_t>(GetHandleFromAddress(ShmPtr(ptr).UserPtr(), fSegmentId)),
MsgDebug(getpid(), size, std::chrono::system_clock::now().time_since_epoch().count()) MsgDebug(getpid(), size, std::chrono::system_clock::now().time_since_epoch().count())
); );
#endif #endif
} }
return ptr; return ShmPtr(ptr);
} }
void Deallocate(boost::interprocess::managed_shared_memory::handle_t handle, uint16_t segmentId) void Deallocate(boost::interprocess::managed_shared_memory::handle_t handle, uint16_t segmentId)

26
fairmq/shmem/Message.h
View File

@ -187,7 +187,8 @@ class Message final : public fair::mq::Message
if (fMeta.fRegionId == 0) { if (fMeta.fRegionId == 0) {
if (fMeta.fSize > 0) { if (fMeta.fSize > 0) {
fManager.GetSegment(fMeta.fSegmentId); fManager.GetSegment(fMeta.fSegmentId);
fLocalPtr = reinterpret_cast<char*>(fManager.GetAddressFromHandle(fMeta.fHandle, fMeta.fSegmentId)); ShmPtr shmPtr(reinterpret_cast<char*>(fManager.GetAddressFromHandle(fMeta.fHandle, fMeta.fSegmentId)));
fLocalPtr = shmPtr.UserPtr();
} else { } else {
fLocalPtr = nullptr; fLocalPtr = nullptr;
} }
@ -202,7 +203,7 @@ class Message final : public fair::mq::Message
} }
} }
return fLocalPtr; return static_cast<void*>(fLocalPtr);
} }
size_t GetSize() const override { return fMeta.fSize; } size_t GetSize() const override { return fMeta.fSize; }
@ -217,7 +218,8 @@ class Message final : public fair::mq::Message
} else if (newSize <= fMeta.fSize) { } else if (newSize <= fMeta.fSize) {
try { try {
try { try {
fLocalPtr = fManager.ShrinkInPlace(newSize, fLocalPtr, fMeta.fSegmentId); ShmPtr shmPtr(fManager.ShrinkInPlace(newSize, static_cast<char*>(fManager.GetAddressFromHandle(fMeta.fHandle, fMeta.fSegmentId)), fMeta.fSegmentId));
fLocalPtr = shmPtr.UserPtr();
fMeta.fSize = newSize; fMeta.fSize = newSize;
return true; return true;
} catch (boost::interprocess::bad_alloc& e) { } catch (boost::interprocess::bad_alloc& e) {
@ -225,12 +227,13 @@ class Message final : public fair::mq::Message
// unused size >= 1000000 bytes: reallocate fully // unused size >= 1000000 bytes: reallocate fully
// unused size < 1000000 bytes: simply reset the size and keep the rest of the buffer until message destruction // unused size < 1000000 bytes: simply reset the size and keep the rest of the buffer until message destruction
if (fMeta.fSize - newSize >= 1000000) { if (fMeta.fSize - newSize >= 1000000) {
char* newPtr = fManager.Allocate(newSize, fAlignment); ShmPtr shmPtr = fManager.Allocate(newSize, fAlignment);
if (newPtr) { if (shmPtr.RealPtr()) {
std::memcpy(newPtr, fLocalPtr, newSize); char* userPtr = shmPtr.UserPtr();
std::memcpy(userPtr, fLocalPtr, newSize);
fManager.Deallocate(fMeta.fHandle, fMeta.fSegmentId); fManager.Deallocate(fMeta.fHandle, fMeta.fSegmentId);
fLocalPtr = newPtr; fLocalPtr = userPtr;
fMeta.fHandle = fManager.GetHandleFromAddress(fLocalPtr, fMeta.fSegmentId); fMeta.fHandle = fManager.GetHandleFromAddress(shmPtr.RealPtr(), fMeta.fSegmentId);
} else { } else {
LOG(debug) << "could not set used size: " << e.what(); LOG(debug) << "could not set used size: " << e.what();
return false; return false;
@ -288,10 +291,11 @@ class Message final : public fair::mq::Message
char* InitializeChunk(const size_t size, size_t alignment = 0) char* InitializeChunk(const size_t size, size_t alignment = 0)
{ {
fLocalPtr = fManager.Allocate(size, alignment); ShmPtr shmPtr = fManager.Allocate(size, alignment);
if (fLocalPtr) { if (shmPtr.RealPtr()) {
fMeta.fHandle = fManager.GetHandleFromAddress(fLocalPtr, fMeta.fSegmentId); fMeta.fHandle = fManager.GetHandleFromAddress(shmPtr.RealPtr(), fMeta.fSegmentId);
fMeta.fSize = size; fMeta.fSize = size;
fLocalPtr = shmPtr.UserPtr();
} }
return fLocalPtr; return fLocalPtr;
} }