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

Shm: Use MakeShmName to construct shm object names

Browse Source
This commit is contained in:
Alexey Rybalchenko
2023-09-28 14:53:36 +02:00
parent 4310d07ed1
commit 8d28824489
5 changed files with 51 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
fairmq/shmem/Common.h
View File

@@ -44,8 +44,12 @@ using RBTreeBestFitSegment = boost::interprocess::basic_managed_shared_memory<ch
boost::interprocess::null_index>;
// boost::interprocess::iset_index>;
inline std::string MakeShmName(const std::string& shmId, const std::string& type) {
return std::string("fmq_" + shmId + "_" + type);
}
inline std::string MakeShmName(const std::string& shmId, const std::string& type, int index) {
return std::string("fmq_" + shmId + "_" + type + "_" + std::to_string(index));
return std::string(MakeShmName(shmId, type) + "_" + std::to_string(index));
}
struct RefCount

12
fairmq/shmem/Manager.h
View File

@@ -58,7 +58,7 @@ class Manager
: fShmId64(config ? config->GetProperty<uint64_t>("shmid", makeShmIdUint64(sessionName)) : makeShmIdUint64(sessionName))
, fShmId(makeShmIdStr(fShmId64))
, fSegmentId(config ? config->GetProperty<uint16_t>("shm-segment-id", 0) : 0)
, fManagementSegment(boost::interprocess::open_or_create, std::string("fmq_" + fShmId + "_mng").c_str(), kManagementSegmentSize)
, fManagementSegment(boost::interprocess::open_or_create, MakeShmName(fShmId, "mng").c_str(), kManagementSegmentSize)
, fShmVoidAlloc(fManagementSegment.get_segment_manager())
, fShmMtx(fManagementSegment.find_or_construct<boost::interprocess::interprocess_mutex>(boost::interprocess::unique_instance)())
, fNumObservedEvents(0)
@@ -158,7 +158,7 @@ class Manager
bool createdSegment = false;
try {
std::string segmentName("fmq_" + fShmId + "_m_" + std::to_string(fSegmentId));
std::string segmentName = MakeShmName(fShmId, "m", fSegmentId);
auto it = fShmSegments->find(fSegmentId);
if (it == fShmSegments->end()) {
// no segment with given id exists, creating
@@ -256,7 +256,7 @@ class Manager
{
using namespace boost::interprocess;
try {
named_mutex monitorStatus(open_only, std::string("fmq_" + id + "_ms").c_str());
named_mutex monitorStatus(open_only, MakeShmName(id, "ms").c_str());
LOG(debug) << "Found fairmq-shmmonitor for shared memory id " << id;
} catch (interprocess_exception&) {
LOG(debug) << "no fairmq-shmmonitor found for shared memory id " << id << ", starting...";
@@ -265,7 +265,7 @@ class Manager
int numTries = 0;
do {
try {
named_mutex monitorStatus(open_only, std::string("fmq_" + id + "_ms").c_str());
named_mutex monitorStatus(open_only, MakeShmName(id, "ms").c_str());
LOG(debug) << "Started fairmq-shmmonitor for shared memory id " << id;
break;
} catch (interprocess_exception&) {
@@ -645,9 +645,9 @@ class Manager
using namespace boost::interprocess;
if (segmentInfo.fAllocationAlgorithm == AllocationAlgorithm::rbtree_best_fit) {
fSegments.emplace(id, RBTreeBestFitSegment(open_only, std::string("fmq_" + fShmId + "_m_" + std::to_string(id)).c_str()));
fSegments.emplace(id, RBTreeBestFitSegment(open_only, MakeShmName(fShmId, "m", id).c_str()));
} else {
fSegments.emplace(id, SimpleSeqFitSegment(open_only, std::string("fmq_" + fShmId + "_m_" + std::to_string(id)).c_str()));
fSegments.emplace(id, SimpleSeqFitSegment(open_only, MakeShmName(fShmId, "m", id).c_str()));
}
} catch (std::out_of_range& oor) {
LOG(error) << "Could not get segment with id '" << id << "': " << oor.what();

70
fairmq/shmem/Monitor.cxx
View File

@@ -75,7 +75,7 @@ Monitor::Monitor(string shmId, bool selfDestruct, bool interactive, bool viewOnl
{
if (!fViewOnly) {
try {
bipc::named_mutex monitorStatus(bipc::create_only, string("fmq_" + fShmId + "_ms").c_str());
bipc::named_mutex monitorStatus(bipc::create_only, MakeShmName(fShmId, "ms").c_str());
} catch (bie&) {
if (fInteractive) {
LOG(error) << "fairmq-shmmonitor for shm id " << fShmId << " is already running. Try `fairmq-shmmonitor --cleanup --shmid " << fShmId << "`, or run in view-only mode (-v)";
@@ -133,7 +133,7 @@ void Monitor::Watch()
using namespace boost::interprocess;
try {
managed_shared_memory managementSegment(open_read_only, std::string("fmq_" + fShmId + "_mng").c_str());
managed_shared_memory managementSegment(open_read_only, MakeShmName(fShmId, "mng").c_str());
fSeenOnce = true;
@@ -181,7 +181,7 @@ bool Monitor::PrintShm(const ShmId& shmId)
using namespace boost::interprocess;
try {
managed_shared_memory managementSegment(open_read_only, std::string("fmq_" + shmId.shmId + "_mng").c_str());
managed_shared_memory managementSegment(open_read_only, MakeShmName(shmId.shmId, "mng").c_str());
VoidAlloc allocInstance(managementSegment.get_segment_manager());
Uint16SegmentInfoHashMap* shmSegments = managementSegment.find<Uint16SegmentInfoHashMap>(unique_instance).first;
@@ -200,9 +200,9 @@ bool Monitor::PrintShm(const ShmId& shmId)
for (const auto& s : *shmSegments) {
if (s.second.fAllocationAlgorithm == AllocationAlgorithm::rbtree_best_fit) {
segments.emplace(s.first, RBTreeBestFitSegment(open_read_only, std::string("fmq_" + shmId.shmId + "_m_" + to_string(s.first)).c_str()));
segments.emplace(s.first, RBTreeBestFitSegment(open_read_only, MakeShmName(shmId.shmId, "m", s.first).c_str()));
} else {
segments.emplace(s.first, SimpleSeqFitSegment(open_read_only, std::string("fmq_" + shmId.shmId + "_m_" + to_string(s.first)).c_str()));
segments.emplace(s.first, SimpleSeqFitSegment(open_read_only, MakeShmName(shmId.shmId, "m", s.first).c_str()));
}
}
@@ -273,7 +273,7 @@ bool Monitor::PrintShm(const ShmId& shmId)
ss << ", size: " << info.fSize;
try {
managed_shared_memory rcCountSegment(open_read_only, std::string("fmq_" + shmId.shmId + "_rrc_" + to_string(id)).c_str());
managed_shared_memory rcCountSegment(open_read_only, MakeShmName(shmId.shmId, "rrc", id).c_str());
ss << ", rcCountSegment size: " << rcCountSegment.get_size();
} catch (bie&) {
ss << ", rcCountSegment: not found";
@@ -333,7 +333,7 @@ void Monitor::CheckHeartbeats()
while (!fTerminating) {
std::this_thread::sleep_for(std::chrono::milliseconds(200));
try {
managed_shared_memory managementSegment(open_read_only, std::string("fmq_" + fShmId + "_mng").c_str());
managed_shared_memory managementSegment(open_read_only, MakeShmName(fShmId, "mng").c_str());
Heartbeat* hb = managementSegment.find<Heartbeat>(unique_instance).first;
if (hb) {
@@ -415,7 +415,7 @@ void Monitor::Interactive()
void Monitor::PrintDebugInfo(const ShmId& shmId __attribute__((unused)))
{
#ifdef FAIRMQ_DEBUG_MODE
string managementSegmentName("fmq_" + shmId.shmId + "_mng");
string managementSegmentName = MakeShmName(shmId.shmId, "mng");
try {
bipc::managed_shared_memory managementSegment(bipc::open_only, managementSegmentName.c_str());
bipc::interprocess_mutex* mtx(managementSegment.find_or_construct<bipc::interprocess_mutex>(bipc::unique_instance)());
@@ -467,7 +467,7 @@ unordered_map<uint16_t, std::vector<BufferDebugInfo>> Monitor::GetDebugInfo(cons
unordered_map<uint16_t, std::vector<BufferDebugInfo>> result;
#ifdef FAIRMQ_DEBUG_MODE
string managementSegmentName("fmq_" + shmId.shmId + "_mng");
string managementSegmentName = MakeShmName(shmId.shmId, "mng");
try {
bipc::managed_shared_memory managementSegment(bipc::open_only, managementSegmentName.c_str());
bipc::interprocess_mutex* mtx(managementSegment.find_or_construct<bipc::interprocess_mutex>(bipc::unique_instance)());
@@ -507,7 +507,7 @@ unsigned long Monitor::GetFreeMemory(const ShmId& shmId, uint16_t segmentId)
{
using namespace boost::interprocess;
try {
bipc::managed_shared_memory managementSegment(bipc::open_only, std::string("fmq_" + shmId.shmId + "_mng").c_str());
bipc::managed_shared_memory managementSegment(bipc::open_only, MakeShmName(shmId.shmId, "mng").c_str());
boost::interprocess::interprocess_mutex* mtx(managementSegment.find_or_construct<bipc::interprocess_mutex>(bipc::unique_instance)());
boost::interprocess::scoped_lock<bipc::interprocess_mutex> lock(*mtx);
@@ -521,10 +521,10 @@ unsigned long Monitor::GetFreeMemory(const ShmId& shmId, uint16_t segmentId)
auto it = shmSegments->find(segmentId);
if (it != shmSegments->end()) {
if (it->second.fAllocationAlgorithm == AllocationAlgorithm::rbtree_best_fit) {
RBTreeBestFitSegment segment(open_read_only, std::string("fmq_" + shmId.shmId + "_m_" + std::to_string(segmentId)).c_str());
RBTreeBestFitSegment segment(open_read_only, MakeShmName(shmId.shmId, "m", segmentId).c_str());
return segment.get_free_memory();
} else {
SimpleSeqFitSegment segment(open_read_only, std::string("fmq_" + shmId.shmId + "_m_" + std::to_string(segmentId)).c_str());
SimpleSeqFitSegment segment(open_read_only, MakeShmName(shmId.shmId, "m", segmentId).c_str());
return segment.get_free_memory();
}
} else {
@@ -546,7 +546,7 @@ bool Monitor::SegmentIsPresent(const ShmId& shmId, uint16_t segmentId)
{
using namespace boost::interprocess;
try {
bipc::managed_shared_memory managementSegment(bipc::open_read_only, std::string("fmq_" + shmId.shmId + "_mng").c_str());
bipc::managed_shared_memory managementSegment(bipc::open_read_only, MakeShmName(shmId.shmId, "mng").c_str());
Uint16SegmentInfoHashMap* shmSegments = managementSegment.find<Uint16SegmentInfoHashMap>(unique_instance).first;
if (!shmSegments) {
@@ -558,9 +558,9 @@ bool Monitor::SegmentIsPresent(const ShmId& shmId, uint16_t segmentId)
if (it != shmSegments->end()) {
try {
if (it->second.fAllocationAlgorithm == AllocationAlgorithm::rbtree_best_fit) {
RBTreeBestFitSegment segment(open_read_only, std::string("fmq_" + shmId.shmId + "_m_" + std::to_string(segmentId)).c_str());
RBTreeBestFitSegment segment(open_read_only, MakeShmName(shmId.shmId, "m", segmentId).c_str());
} else {
SimpleSeqFitSegment segment(open_read_only, std::string("fmq_" + shmId.shmId + "_m_" + std::to_string(segmentId)).c_str());
SimpleSeqFitSegment segment(open_read_only, MakeShmName(shmId.shmId, "m", segmentId).c_str());
}
} catch (bie&) {
LOG(error) << "Could not find segment with id '" << segmentId << "' for shmId '" << shmId.shmId << "'";
@@ -587,7 +587,7 @@ bool Monitor::RegionIsPresent(const ShmId& shmId, uint16_t regionId)
{
using namespace boost::interprocess;
try {
bipc::managed_shared_memory managementSegment(bipc::open_read_only, std::string("fmq_" + shmId.shmId + "_mng").c_str());
bipc::managed_shared_memory managementSegment(bipc::open_read_only, MakeShmName(shmId.shmId, "mng").c_str());
Uint16RegionInfoHashMap* shmRegions = managementSegment.find<Uint16RegionInfoHashMap>(bipc::unique_instance).first;
if (!shmRegions) {
@@ -595,7 +595,7 @@ bool Monitor::RegionIsPresent(const ShmId& shmId, uint16_t regionId)
return false;
}
std::string regionFileName("fmq_" + shmId.shmId + "_rg_" + to_string(regionId));
std::string regionFileName(MakeShmName(shmId.shmId, "rg", regionId));
auto it = shmRegions->find(regionId);
if (it != shmRegions->end()) {
@@ -663,7 +663,7 @@ std::vector<std::pair<std::string, bool>> Monitor::Cleanup(const ShmId& shmIdT,
LOG(info) << "Cleaning up for shared memory id '" << shmId << "'...";
}
string managementSegmentName("fmq_" + shmId + "_mng");
string managementSegmentName = MakeShmName(shmId, "mng");
try {
bipc::managed_shared_memory managementSegment(bipc::open_read_only, managementSegmentName.c_str());
@@ -681,12 +681,12 @@ std::vector<std::pair<std::string, bool>> Monitor::Cleanup(const ShmId& shmIdT,
LOG(info) << "Found UnmanagedRegion with id: " << id << ", path: '" << path << "', flags: " << flags << ", fDestroyed: " << info.fDestroyed << ".";
}
if (!path.empty()) {
result.emplace_back(Remove<bipc::file_mapping>(path + "fmq_" + shmId + "_rg_" + to_string(id), verbose));
result.emplace_back(Remove<bipc::file_mapping>(path + MakeShmName(shmId, "rg", id), verbose));
} else {
result.emplace_back(Remove<bipc::shared_memory_object>("fmq_" + shmId + "_rg_" + to_string(id), verbose));
result.emplace_back(Remove<bipc::shared_memory_object>(MakeShmName(shmId, "rg", id), verbose));
}
result.emplace_back(Remove<bipc::message_queue>("fmq_" + shmId + "_rgq_" + to_string(id), verbose));
result.emplace_back(Remove<bipc::shared_memory_object>("fmq_" + shmId + "_rrc_" + to_string(id), verbose));
result.emplace_back(Remove<bipc::message_queue>(MakeShmName(shmId, "rgq", id), verbose));
result.emplace_back(Remove<bipc::shared_memory_object>(MakeShmName(shmId, "rrc", id), verbose));
}
}
@@ -696,7 +696,7 @@ std::vector<std::pair<std::string, bool>> Monitor::Cleanup(const ShmId& shmIdT,
LOG(info) << "Found " << shmSegments->size() << " managed segments...";
}
for (const auto& segment : *shmSegments) {
result.emplace_back(Remove<bipc::shared_memory_object>("fmq_" + shmId + "_m_" + to_string(segment.first), verbose));
result.emplace_back(Remove<bipc::shared_memory_object>(MakeShmName(shmId, "m", segment.first), verbose));
}
} else {
if (verbose) {
@@ -726,7 +726,7 @@ std::vector<std::pair<std::string, bool>> Monitor::Cleanup(const SessionId& sess
std::vector<std::pair<std::string, bool>> Monitor::CleanupFull(const ShmId& shmId, bool verbose /* = true */)
{
auto result = Cleanup(shmId, verbose);
result.emplace_back(Remove<bipc::named_mutex>("fmq_" + shmId.shmId + "_ms", verbose));
result.emplace_back(Remove<bipc::named_mutex>(MakeShmName(shmId.shmId, "ms"), verbose));
return result;
}
@@ -746,7 +746,7 @@ void Monitor::ResetContent(const ShmId& shmIdT, bool verbose /* = true */)
cout << "Resetting segments content for shared memory id '" << shmId << "'..." << endl;
}
string managementSegmentName("fmq_" + shmId + "_mng");
string managementSegmentName = MakeShmName(shmId, "mng");
try {
using namespace boost::interprocess;
managed_shared_memory managementSegment(open_only, managementSegmentName.c_str());
@@ -754,18 +754,18 @@ void Monitor::ResetContent(const ShmId& shmIdT, bool verbose /* = true */)
Uint16SegmentInfoHashMap* segmentInfos = managementSegment.find<Uint16SegmentInfoHashMap>(unique_instance).first;
if (segmentInfos) {
cout << "Found info for " << segmentInfos->size() << " managed segments" << endl;
for (const auto& s : *segmentInfos) {
for (const auto& [id, info] : *segmentInfos) {
if (verbose) {
cout << "Resetting content of segment '" << "fmq_" << shmId << "_m_" << s.first << "'..." << endl;
cout << "Resetting content of segment '" << MakeShmName(shmId, "m", id) << "'..." << endl;
}
try {
if (s.second.fAllocationAlgorithm == AllocationAlgorithm::rbtree_best_fit) {
RBTreeBestFitSegment segment(open_only, std::string("fmq_" + shmId + "_m_" + to_string(s.first)).c_str());
if (info.fAllocationAlgorithm == AllocationAlgorithm::rbtree_best_fit) {
RBTreeBestFitSegment segment(open_only, MakeShmName(shmId, "m", id).c_str());
void* ptr = segment.get_segment_manager();
size_t size = segment.get_segment_manager()->get_size();
new(ptr) segment_manager<char, rbtree_best_fit<mutex_family, offset_ptr<void>>, null_index>(size);
} else {
SimpleSeqFitSegment segment(open_only, std::string("fmq_" + shmId + "_m_" + to_string(s.first)).c_str());
SimpleSeqFitSegment segment(open_only, MakeShmName(shmId, "m", id).c_str());
void* ptr = segment.get_segment_manager();
size_t size = segment.get_segment_manager()->get_size();
new(ptr) segment_manager<char, simple_seq_fit<mutex_family, offset_ptr<void>>, null_index>(size);
@@ -775,7 +775,7 @@ void Monitor::ResetContent(const ShmId& shmIdT, bool verbose /* = true */)
}
} catch (bie& e) {
if (verbose) {
cout << "Error resetting content of segment '" << std::string("fmq_" + shmId + "_m_" + to_string(s.first)) << "': " << e.what() << endl;
cout << "Error resetting content of segment '" << MakeShmName(shmId, "m", id) << "': " << e.what() << endl;
}
}
}
@@ -787,8 +787,8 @@ void Monitor::ResetContent(const ShmId& shmIdT, bool verbose /* = true */)
if (shmRegions) {
for (const auto& region : *shmRegions) {
uint16_t id = region.first;
Remove<bipc::message_queue>("fmq_" + shmId + "_rgq_" + to_string(id), verbose);
Remove<bipc::shared_memory_object>("fmq_" + shmId + "_rrc_" + to_string(id), verbose);
Remove<bipc::message_queue>(MakeShmName(shmId, "rgq", id), verbose);
Remove<bipc::shared_memory_object>(MakeShmName(shmId, "rrc", id), verbose);
}
}
} catch (bie& e) {
@@ -817,7 +817,7 @@ void Monitor::ResetContent(const ShmId& shmIdT, const std::vector<SegmentConfig>
using namespace boost::interprocess;
std::string shmId = shmIdT.shmId;
std::string managementSegmentName("fmq_" + shmId + "_mng");
std::string managementSegmentName = MakeShmName(shmId, "mng");
// delete management segment
cout << "deleting management segment" << endl;
Remove<bipc::shared_memory_object>(managementSegmentName, verbose);
@@ -865,7 +865,7 @@ Monitor::~Monitor()
Cleanup(ShmId{fShmId});
}
if (!fViewOnly) {
RemoveMutex("fmq_" + fShmId + "_ms");
RemoveMutex(MakeShmName(fShmId, "ms"));
}
}

12
fairmq/shmem/Segment.h
View File

@@ -28,17 +28,13 @@ struct Segment
friend class Monitor;
Segment(const std::string& shmId, uint16_t id, size_t size, SimpleSeqFit)
: fSegment(SimpleSeqFitSegment(boost::interprocess::open_or_create,
std::string("fmq_" + shmId + "_m_" + std::to_string(id)).c_str(),
size))
: fSegment(SimpleSeqFitSegment(boost::interprocess::open_or_create, MakeShmName(shmId, "m", id).c_str(), size))
{
Register(shmId, id, AllocationAlgorithm::simple_seq_fit);
}
Segment(const std::string& shmId, uint16_t id, size_t size, RBTreeBestFit)
: fSegment(RBTreeBestFitSegment(boost::interprocess::open_or_create,
std::string("fmq_" + shmId + "_m_" + std::to_string(id)).c_str(),
size))
: fSegment(RBTreeBestFitSegment(boost::interprocess::open_or_create, MakeShmName(shmId, "m", id).c_str(), size))
{
Register(shmId, id, AllocationAlgorithm::rbtree_best_fit);
}
@@ -58,7 +54,7 @@ struct Segment
static void Remove(const std::string& shmId, uint16_t id)
{
Monitor::RemoveObject("fmq_" + shmId + "_m_" + std::to_string(id));
Monitor::RemoveObject(MakeShmName(shmId, "m", id));
}
private:
@@ -67,7 +63,7 @@ struct Segment
static void Register(const std::string& shmId, uint16_t id, AllocationAlgorithm allocAlgo)
{
using namespace boost::interprocess;
managed_shared_memory mngSegment(open_or_create, std::string("fmq_" + shmId + "_mng").c_str(), kManagementSegmentSize);
managed_shared_memory mngSegment(open_or_create, MakeShmName(shmId, "mng").c_str(), kManagementSegmentSize);
VoidAlloc alloc(mngSegment.get_segment_manager());
Uint16SegmentInfoHashMap* shmSegments = mngSegment.find_or_construct<Uint16SegmentInfoHashMap>(unique_instance)(alloc);

6
fairmq/shmem/UnmanagedRegion.h
View File

@@ -273,10 +273,6 @@ struct UnmanagedRegion
RegionCallback fCallback;
RegionBulkCallback fBulkCallback;
static std::string MakeSegmentName(const std::string& shmId, std::string_view segment, int regionIndex) {
return tools::ToString("fmq_", shmId, "_", segment, "_", regionIndex);
}
static RegionConfig makeRegionConfig(uint16_t id)
{
RegionConfig regionCfg;
@@ -288,7 +284,7 @@ struct UnmanagedRegion
{
using namespace boost::interprocess;
LOG(debug) << "Registering unmanaged shared memory region with id " << cfg.id.value();
managed_shared_memory mngSegment(open_or_create, std::string("fmq_" + shmId + "_mng").c_str(), kManagementSegmentSize);
managed_shared_memory mngSegment(open_or_create, MakeShmName(shmId, "mng").c_str(), kManagementSegmentSize);
VoidAlloc alloc(mngSegment.get_segment_manager());
Uint16RegionInfoHashMap* shmRegions = mngSegment.find_or_construct<Uint16RegionInfoHashMap>(unique_instance)(alloc);