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

shm: fix initialization of rc segment when region is created externally

Browse Source
This commit is contained in:
Alexey Rybalchenko 2023-11-24 12:12:20 +01:00
parent ff1f9b94ef
commit faf577086a
2 changed files with 10 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
fairmq/shmem/Manager.h
View File

@ -323,7 +323,6 @@ class Manager
} }
const uint16_t id = cfg.id.value(); const uint16_t id = cfg.id.value();
const uint64_t rcSegmentSize = cfg.rcSegmentSize;
std::lock_guard<std::mutex> lock(fLocalRegionsMtx); std::lock_guard<std::mutex> lock(fLocalRegionsMtx);
@ -340,6 +339,12 @@ class Manager
LOG(debug) << "Unmanaged region (view) already present, promoting to controller"; LOG(debug) << "Unmanaged region (view) already present, promoting to controller";
region->BecomeController(cfg); region->BecomeController(cfg);
} else { } else {
// we need to update local config, if the region information already exists
auto info = fShmRegions->find(id);
if (info != fShmRegions->end()) {
cfg.rcSegmentSize = info->second.fRCSegmentSize;
}
auto res = fRegions.emplace(id, std::make_unique<UnmanagedRegion>(fShmId, size, true, cfg)); auto res = fRegions.emplace(id, std::make_unique<UnmanagedRegion>(fShmId, size, true, cfg));
region = res.first->second.get(); region = res.first->second.get();
} }
@ -348,7 +353,6 @@ class Manager
// start ack receiver only if a callback has been provided. // start ack receiver only if a callback has been provided.
if (callback || bulkCallback) { if (callback || bulkCallback) {
region->SetCallbacks(callback, bulkCallback); region->SetCallbacks(callback, bulkCallback);
region->InitializeRefCountSegment(rcSegmentSize);
region->InitializeQueues(); region->InitializeQueues();
region->StartAckSender(); region->StartAckSender();
region->StartAckReceiver(); region->StartAckReceiver();
@ -401,19 +405,18 @@ class Manager
try { try {
RegionConfig cfg; RegionConfig cfg;
const uint64_t rcSegmentSize = cfg.rcSegmentSize;
// get region info // get region info
{ {
boost::interprocess::scoped_lock<boost::interprocess::interprocess_mutex> shmLock(*fShmMtx); boost::interprocess::scoped_lock<boost::interprocess::interprocess_mutex> shmLock(*fShmMtx);
RegionInfo regionInfo = fShmRegions->at(id); RegionInfo regionInfo = fShmRegions->at(id);
cfg.id = id; cfg.id = id;
cfg.creationFlags = regionInfo.fCreationFlags; cfg.creationFlags = regionInfo.fCreationFlags;
cfg.rcSegmentSize = regionInfo.fRCSegmentSize;
cfg.path = regionInfo.fPath.c_str(); cfg.path = regionInfo.fPath.c_str();
} }
// LOG(debug) << "Located remote region with id '" << id << "', path: '" << cfg.path << "', flags: '" << cfg.creationFlags << "'"; // LOG(debug) << "Located remote region with id '" << id << "', path: '" << cfg.path << "', flags: '" << cfg.creationFlags << "'";
auto r = fRegions.emplace(id, std::make_unique<UnmanagedRegion>(fShmId, 0, false, std::move(cfg))); auto r = fRegions.emplace(id, std::make_unique<UnmanagedRegion>(fShmId, 0, false, std::move(cfg)));
r.first->second->InitializeRefCountSegment(rcSegmentSize);
r.first->second->InitializeQueues(); r.first->second->InitializeQueues();
r.first->second->StartAckSender(); r.first->second->StartAckSender();
return r.first->second.get(); return r.first->second.get();
@ -482,6 +485,7 @@ class Manager
cfg.id = info.id; cfg.id = info.id;
cfg.creationFlags = regionInfo.fCreationFlags; cfg.creationFlags = regionInfo.fCreationFlags;
cfg.path = regionInfo.fPath.c_str(); cfg.path = regionInfo.fPath.c_str();
cfg.rcSegmentSize = regionInfo.fRCSegmentSize;
regionCfgs.emplace(info.id, cfg); regionCfgs.emplace(info.id, cfg);
// fill the ptr+size info after shmLock is released, to avoid constructing local region under it // fill the ptr+size info after shmLock is released, to avoid constructing local region under it
} else { } else {
@ -503,10 +507,8 @@ class Manager
if (it != fRegions.end()) { if (it != fRegions.end()) {
region = it->second.get(); region = it->second.get();
} else { } else {
const uint64_t rcSegmentSize = cfgIt->second.rcSegmentSize;
auto r = fRegions.emplace(cfgIt->first, std::make_unique<UnmanagedRegion>(fShmId, 0, false, cfgIt->second)); auto r = fRegions.emplace(cfgIt->first, std::make_unique<UnmanagedRegion>(fShmId, 0, false, cfgIt->second));
region = r.first->second.get(); region = r.first->second.get();
region->InitializeRefCountSegment(rcSegmentSize);
region->InitializeQueues(); region->InitializeQueues();
region->StartAckSender(); region->StartAckSender();
} }

2
fairmq/shmem/UnmanagedRegion.h
View File

@ -146,6 +146,8 @@ struct UnmanagedRegion
LOG(debug) << "Successfully zeroed free memory of region " << id << "."; LOG(debug) << "Successfully zeroed free memory of region " << id << ".";
} }
InitializeRefCountSegment(cfg.rcSegmentSize);
if (fControlling && created) { if (fControlling && created) {
Register(shmId, cfg); Register(shmId, cfg);
} }