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

shm: error on duplicate region IDs

Browse Source
This commit is contained in:
Alexey Rybalchenko 2022-10-04 12:15:21 +02:00
parent 58aa2b4f88
commit 42ce691f57
2 changed files with 33 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
fairmq/shmem/Manager.h
View File

@ -396,7 +396,23 @@ class Manager
const uint16_t id = cfg.id.value();
std::lock_guard<std::mutex> lock(fLocalRegionsMtx);
auto& region = fRegions[id] = std::make_unique<UnmanagedRegion>(fShmId, size, true, cfg);
UnmanagedRegion* region = nullptr;
auto it = fRegions.find(id);
if (it != fRegions.end()) {
region = it->second.get();
if (region->fControlling) {
LOG(error) << "Unmanaged Region with id " << id << " already exists. Only unique IDs per session are allowed.";
throw TransportError(tools::ToString("Unmanaged Region with id ", id, " already exists. Only unique IDs per session are allowed."));
}
LOG(debug) << "Unmanaged region (view) already present, promoting to controller";
region->BecomeController(cfg);
} else {
auto res = fRegions.emplace(id, std::make_unique<UnmanagedRegion>(fShmId, size, true, cfg));
region = res.first->second.get();
}
// LOG(debug) << "Created region with id '" << id << "', path: '" << cfg.path << "', flags: '" << cfg.creationFlags << "'";
// start ack receiver only if a callback has been provided.
@ -406,7 +422,7 @@ class Manager
region->StartAckSender();
region->StartAckReceiver();
}
result.first = region.get();
result.first = region;
result.second = id;
}
fRegionsGen += 1; // signal TL cache invalidation

19
fairmq/shmem/UnmanagedRegion.h
View File

@ -146,7 +146,7 @@ struct UnmanagedRegion
LOG(debug) << "Successfully zeroed free memory of region " << id << ".";
}
if (fControlling) {
if (fControlling && created) {
Register(shmId, cfg);
}
@ -160,6 +160,13 @@ struct UnmanagedRegion
UnmanagedRegion& operator=(const UnmanagedRegion&) = delete;
UnmanagedRegion& operator=(UnmanagedRegion&&) = delete;
void BecomeController(RegionConfig& cfg)
{
fControlling = true;
fLinger = cfg.linger;
fRemoveOnDestruction = cfg.removeOnDestruction;
}
void Zero()
{
memset(fRegion.get_address(), 0x00, fRegion.get_size());
@ -263,10 +270,14 @@ struct UnmanagedRegion
EventCounter* eventCounter = mngSegment.find_or_construct<EventCounter>(unique_instance)(0);
bool newShmRegionCreated = shmRegions->emplace(cfg.id.value(), RegionInfo(cfg.path.c_str(), cfg.creationFlags, cfg.userFlags, cfg.size, alloc)).second;
if (newShmRegionCreated) {
(eventCounter->fCount)++;
auto it = shmRegions->find(cfg.id.value());
if (it != shmRegions->end()) {
LOG(error) << "Unmanaged Region with id " << cfg.id.value() << " has already been registered. Only unique IDs per session are allowed.";
throw TransportError(tools::ToString("Unmanaged Region with id ", cfg.id.value(), " has already been registered. Only unique IDs per session are allowed."));
}
shmRegions->emplace(cfg.id.value(), RegionInfo(cfg.path.c_str(), cfg.creationFlags, cfg.userFlags, cfg.size, alloc)).second;
(eventCounter->fCount)++;
}
void SetCallbacks(RegionCallback callback, RegionBulkCallback bulkCallback)