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

use thread local cache to avoid interprocess lock on shm GetData

Browse Source
This commit is contained in:
Gvozden Neskovic 2021-05-06 11:04:01 +02:00 committed by Alexey Rybalchenko
parent a90dbf64de
commit f7ba3052aa
1 changed files with 30 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
fairmq/shmem/Manager.h
View File

@ -314,6 +314,7 @@ class Manager
(fEventCounter->fCount)++; (fEventCounter->fCount)++;
} }
fRegionsGen += 1; // signal TL cache invalidation
fRegionEventsCV.notify_all(); fRegionEventsCV.notify_all();
return result; return result;
@ -327,8 +328,28 @@ class Manager
Region* GetRegion(const uint16_t id) Region* GetRegion(const uint16_t id)
{ {
// NOTE: gcc optimizations. Prevent loading tls addresses many times in the fast path
const auto &lTlCache = fTlRegionCache;
const auto &lTlCacheVec = lTlCache.fRegionsTLCache;
const auto lTlCacheGen = lTlCache.fRegionsTLCacheGen;
// fast path
for (const auto &lRegion : lTlCacheVec) {
if ((lRegion.second == id) && (lTlCacheGen == fRegionsGen)) {
return lRegion.first;
}
}
boost::interprocess::scoped_lock<boost::interprocess::named_mutex> lock(fShmMtx); boost::interprocess::scoped_lock<boost::interprocess::named_mutex> lock(fShmMtx);
return GetRegionUnsafe(id); // slow path: check invalidation
if (lTlCacheGen != fRegionsGen) {
fTlRegionCache.fRegionsTLCache.clear();
}
auto *lRegion = GetRegionUnsafe(id);
fTlRegionCache.fRegionsTLCache.emplace_back(std::make_pair(lRegion, id));
fTlRegionCache.fRegionsTLCacheGen = fRegionsGen;
return lRegion;
} }
Region* GetRegionUnsafe(const uint16_t id) Region* GetRegionUnsafe(const uint16_t id)
@ -366,6 +387,7 @@ class Manager
fShmRegions->at(id).fDestroyed = true; fShmRegions->at(id).fDestroyed = true;
(fEventCounter->fCount)++; (fEventCounter->fCount)++;
} }
fRegionsGen += 1; // signal TL cache invalidation
fRegionEventsCV.notify_all(); fRegionEventsCV.notify_all();
} }
@ -613,6 +635,7 @@ class Manager
using namespace boost::interprocess; using namespace boost::interprocess;
bool lastRemoved = false; bool lastRemoved = false;
fRegionsGen += 1; // signal TL cache invalidation
UnsubscribeFromRegionEvents(); UnsubscribeFromRegionEvents();
{ {
@ -665,6 +688,12 @@ class Manager
Uint16SegmentInfoHashMap* fShmSegments; Uint16SegmentInfoHashMap* fShmSegments;
Uint16RegionInfoHashMap* fShmRegions; Uint16RegionInfoHashMap* fShmRegions;
std::unordered_map<uint16_t, std::unique_ptr<Region>> fRegions; std::unordered_map<uint16_t, std::unique_ptr<Region>> fRegions;
// make sure this is alone in the cache line: mostly read
alignas(128) inline static std::atomic<unsigned long> fRegionsGen = 0ul;
inline static thread_local struct ManagerTLCache {
unsigned long fRegionsTLCacheGen;
std::vector<std::pair<Region*, uint16_t>> fRegionsTLCache;
} fTlRegionCache;
std::atomic<bool> fInterrupted; std::atomic<bool> fInterrupted;
std::atomic<int32_t> fMsgCounter; // TODO: find a better lifetime solution instead of the counter std::atomic<int32_t> fMsgCounter; // TODO: find a better lifetime solution instead of the counter