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

extend region config

Browse Source
This commit is contained in:
Alexey Rybalchenko 2021-12-06 23:44:38 +01:00 committed by Dennis Klein
parent eef42d2dea
commit 80ed45df63
7 changed files with 156 additions and 90 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
fairmq/TransportFactory.h
View File

@ -145,7 +145,6 @@ class TransportFactory
int flags = 0, int flags = 0,
RegionConfig cfg = RegionConfig()) = 0; RegionConfig cfg = RegionConfig()) = 0;
/// @brief Create new UnmanagedRegion /// @brief Create new UnmanagedRegion
/// @param size size of the region /// @param size size of the region
/// @param callback callback to be called when a message belonging to this region is no longer needed by the transport /// @param callback callback to be called when a message belonging to this region is no longer needed by the transport

11
fairmq/UnmanagedRegion.h
View File

@ -16,6 +16,7 @@
#include <functional> // std::function #include <functional> // std::function
#include <memory> // std::unique_ptr #include <memory> // std::unique_ptr
#include <optional> // std::optional
#include <ostream> #include <ostream>
#include <string> #include <string>
#include <vector> #include <vector>
@ -118,6 +119,13 @@ inline std::ostream& operator<<(std::ostream& os, const RegionEvent& event)
} }
} }
enum class RegionConstruction : int
{
open,
create,
open_or_create
};
struct RegionConfig struct RegionConfig
{ {
RegionConfig() = default; RegionConfig() = default;
@ -129,9 +137,12 @@ struct RegionConfig
bool lock = false; /// mlock region after creation bool lock = false; /// mlock region after creation
bool zero = false; /// zero region content after creation bool zero = false; /// zero region content after creation
bool removeOnDestruction = true; /// remove the region on object destruction
RegionConstruction constructionMode = RegionConstruction::create; /// how to construct the region: create/open/open_or_create
int creationFlags = 0; /// flags passed to the underlying transport on region creation int creationFlags = 0; /// flags passed to the underlying transport on region creation
int64_t userFlags = 0; /// custom flags that have no effect on the transport, but can be retrieved from the region by the user int64_t userFlags = 0; /// custom flags that have no effect on the transport, but can be retrieved from the region by the user
std::string path = ""; /// file path, if the region is backed by a file std::string path = ""; /// file path, if the region is backed by a file
std::optional<uint16_t> id = std::nullopt; /// region id
uint32_t linger = 100; /// delay in ms before region destruction to collect outstanding events uint32_t linger = 100; /// delay in ms before region destruction to collect outstanding events
}; };

92
fairmq/shmem/Manager.h
View File

@ -234,15 +234,16 @@ class Manager
std::string op("create/open"); std::string op("create/open");
try { try {
std::string segmentName("fmq_" + fShmId + "_m_" + std::to_string(fSegmentId));
auto it = fShmSegments->find(fSegmentId); auto it = fShmSegments->find(fSegmentId);
if (it == fShmSegments->end()) { if (it == fShmSegments->end()) {
op = "create"; op = "create";
// no segment with given id exists, creating // no segment with given id exists, creating
if (allocationAlgorithm == "rbtree_best_fit") { if (allocationAlgorithm == "rbtree_best_fit") {
fSegments.emplace(fSegmentId, RBTreeBestFitSegment(create_only, std::string("fmq_" + fShmId + "_m_" + std::to_string(fSegmentId)).c_str(), size)); fSegments.emplace(fSegmentId, RBTreeBestFitSegment(create_only, segmentName.c_str(), size));
fShmSegments->emplace(fSegmentId, AllocationAlgorithm::rbtree_best_fit); fShmSegments->emplace(fSegmentId, AllocationAlgorithm::rbtree_best_fit);
} else if (allocationAlgorithm == "simple_seq_fit") { } else if (allocationAlgorithm == "simple_seq_fit") {
fSegments.emplace(fSegmentId, SimpleSeqFitSegment(create_only, std::string("fmq_" + fShmId + "_m_" + std::to_string(fSegmentId)).c_str(), size)); fSegments.emplace(fSegmentId, SimpleSeqFitSegment(create_only, segmentName.c_str(), size));
fShmSegments->emplace(fSegmentId, AllocationAlgorithm::simple_seq_fit); fShmSegments->emplace(fSegmentId, AllocationAlgorithm::simple_seq_fit);
} }
ss << "Created "; ss << "Created ";
@ -257,13 +258,13 @@ class Manager
op = "open"; op = "open";
// found segment with the given id, opening // found segment with the given id, opening
if (it->second.fAllocationAlgorithm == AllocationAlgorithm::rbtree_best_fit) { if (it->second.fAllocationAlgorithm == AllocationAlgorithm::rbtree_best_fit) {
fSegments.emplace(fSegmentId, RBTreeBestFitSegment(open_only, std::string("fmq_" + fShmId + "_m_" + std::to_string(fSegmentId)).c_str())); fSegments.emplace(fSegmentId, RBTreeBestFitSegment(open_only, segmentName.c_str()));
if (allocationAlgorithm != "rbtree_best_fit") { if (allocationAlgorithm != "rbtree_best_fit") {
LOG(warn) << "Allocation algorithm of the opened segment is rbtree_best_fit, but requested is " << allocationAlgorithm << ". Ignoring requested setting."; LOG(warn) << "Allocation algorithm of the opened segment is rbtree_best_fit, but requested is " << allocationAlgorithm << ". Ignoring requested setting.";
allocationAlgorithm = "rbtree_best_fit"; allocationAlgorithm = "rbtree_best_fit";
} }
} else { } else {
fSegments.emplace(fSegmentId, SimpleSeqFitSegment(open_only, std::string("fmq_" + fShmId + "_m_" + std::to_string(fSegmentId)).c_str())); fSegments.emplace(fSegmentId, SimpleSeqFitSegment(open_only, segmentName.c_str()));
if (allocationAlgorithm != "simple_seq_fit") { if (allocationAlgorithm != "simple_seq_fit") {
LOG(warn) << "Allocation algorithm of the opened segment is simple_seq_fit, but requested is " << allocationAlgorithm << ". Ignoring requested setting."; LOG(warn) << "Allocation algorithm of the opened segment is simple_seq_fit, but requested is " << allocationAlgorithm << ". Ignoring requested setting.";
allocationAlgorithm = "simple_seq_fit"; allocationAlgorithm = "simple_seq_fit";
@ -276,7 +277,7 @@ class Manager
<< " Available: " << boost::apply_visitor(SegmentFreeMemory(), fSegments.at(fSegmentId)) << " bytes." << " Available: " << boost::apply_visitor(SegmentFreeMemory(), fSegments.at(fSegmentId)) << " bytes."
<< " Allocation algorithm: " << allocationAlgorithm; << " Allocation algorithm: " << allocationAlgorithm;
LOG(debug) << ss.str(); LOG(debug) << ss.str();
} catch(interprocess_exception& bie) { } catch (interprocess_exception& bie) {
LOG(error) << "Failed to " << op << " shared memory segment (" << "fmq_" << fShmId << "_m_" << fSegmentId << "): " << bie.what(); LOG(error) << "Failed to " << op << " shared memory segment (" << "fmq_" << fShmId << "_m_" << fSegmentId << "): " << bie.what();
throw TransportError(tools::ToString("Failed to ", op, " shared memory segment (", "fmq_", fShmId, "_m_", fSegmentId, "): ", bie.what())); throw TransportError(tools::ToString("Failed to ", op, " shared memory segment (", "fmq_", fShmId, "_m_", fSegmentId, "): ", bie.what()));
} }
@ -380,50 +381,60 @@ class Manager
{ {
boost::interprocess::scoped_lock<boost::interprocess::named_mutex> lock(fShmMtx); boost::interprocess::scoped_lock<boost::interprocess::named_mutex> lock(fShmMtx);
RegionCounter* rc = fManagementSegment.find<RegionCounter>(unique_instance).first; if (!cfg.id.has_value()) {
RegionCounter* rc = fManagementSegment.find<RegionCounter>(unique_instance).first;
if (rc) { if (rc) {
LOG(debug) << "region counter found, with value of " << rc->fCount << ". incrementing."; LOG(trace) << "region counter found, with value of " << rc->fCount << ". incrementing.";
(rc->fCount)++; (rc->fCount)++;
LOG(debug) << "incremented region counter, now: " << rc->fCount; LOG(trace) << "incremented region counter, now: " << rc->fCount;
} else { } else {
LOG(debug) << "no region counter found, creating one and initializing with 1024"; LOG(trace) << "no region counter found, creating one and initializing with 1024";
rc = fManagementSegment.construct<RegionCounter>(unique_instance)(1024); rc = fManagementSegment.construct<RegionCounter>(unique_instance)(1024);
LOG(debug) << "initialized region counter with: " << rc->fCount; LOG(trace) << "initialized region counter with: " << rc->fCount;
}
cfg.id = rc->fCount;
} else if (cfg.id.value() == 0) {
LOG(error) << "User-given UnmanagedRegion ID must not be 0.";
throw TransportError("User-given UnmanagedRegion ID must not be 0.");
} }
uint16_t id = rc->fCount; auto it = fRegions.find(cfg.id.value());
auto it = fRegions.find(id);
if (it != fRegions.end()) { if (it != fRegions.end()) {
LOG(error) << "Trying to create a region that already exists"; LOG(error) << "Trying to open/create a UnmanagedRegion that already exists (id: " << cfg.id.value() << ")";
return {nullptr, id}; throw TransportError(tools::ToString("Trying to open/create a UnmanagedRegion that already exists (id: ", cfg.id.value(), ")"));
} }
auto r = fRegions.emplace(id, std::make_unique<Region>(fShmId, id, size, false, callback, bulkCallback, cfg)); Region& region = *(fRegions.emplace(cfg.id.value(), std::make_unique<Region>(fShmId, size, false, callback, bulkCallback, cfg)).first->second);
// LOG(debug) << "Created region with id '" << id << "', path: '" << cfg.path << "', flags: '" << cfg.creationFlags << "'"; // LOG(debug) << "Created region with id '" << cfg.id.value() << "', path: '" << cfg.path << "', flags: '" << cfg.creationFlags << "'";
if (cfg.lock) { if (cfg.lock) {
LOG(debug) << "Locking region " << id << "..."; LOG(debug) << "Locking region " << cfg.id.value() << "...";
if (mlock(r.first->second->fRegion.get_address(), r.first->second->fRegion.get_size()) == -1) { if (mlock(region.fRegion.get_address(), region.fRegion.get_size()) == -1) {
LOG(error) << "Could not lock region " << id << ". Code: " << errno << ", reason: " << strerror(errno); LOG(error) << "Could not lock region " << cfg.id.value() << ". Code: " << errno << ", reason: " << strerror(errno);
throw TransportError(tools::ToString("Could not lock region ", id, ": ", strerror(errno))); throw TransportError(tools::ToString("Could not lock region ", cfg.id.value(), ": ", strerror(errno)));
} }
LOG(debug) << "Successfully locked region " << id << "."; LOG(debug) << "Successfully locked region " << cfg.id.value() << ".";
} }
if (cfg.zero) { if (cfg.zero) {
LOG(debug) << "Zeroing free memory of region " << id << "..."; LOG(debug) << "Zeroing free memory of region " << cfg.id.value() << "...";
memset(r.first->second->fRegion.get_address(), 0x00, r.first->second->fRegion.get_size()); memset(region.fRegion.get_address(), 0x00, region.fRegion.get_size());
LOG(debug) << "Successfully zeroed free memory of region " << id << "."; LOG(debug) << "Successfully zeroed free memory of region " << cfg.id.value() << ".";
} }
fShmRegions->emplace(id, RegionInfo(cfg.path.c_str(), cfg.creationFlags, cfg.userFlags, fShmVoidAlloc)); bool newRegionCreated = fShmRegions->emplace(cfg.id.value(), RegionInfo(cfg.path.c_str(), cfg.creationFlags, cfg.userFlags, fShmVoidAlloc)).second;
r.first->second->StartReceivingAcks(); // start ack receiver only if a callback has been provided.
result.first = &(r.first->second->fRegion); if (callback || bulkCallback) {
result.second = id; region.StartAckReceiver();
}
result.first = &(region.fRegion);
result.second = cfg.id.value();
(fEventCounter->fCount)++; if (newRegionCreated) {
(fEventCounter->fCount)++;
}
} }
fRegionsGen += 1; // signal TL cache invalidation fRegionsGen += 1; // signal TL cache invalidation
fRegionEventsShmCV.notify_all(); fRegionEventsShmCV.notify_all();
@ -473,11 +484,12 @@ class Manager
// get region info // get region info
RegionInfo regionInfo = fShmRegions->at(id); RegionInfo regionInfo = fShmRegions->at(id);
RegionConfig cfg; RegionConfig cfg;
cfg.id = id;
cfg.creationFlags = regionInfo.fCreationFlags; cfg.creationFlags = regionInfo.fCreationFlags;
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<Region>(fShmId, id, 0, true, nullptr, nullptr, std::move(cfg))); auto r = fRegions.emplace(id, std::make_unique<Region>(fShmId, 0, true, nullptr, nullptr, std::move(cfg)));
return r.first->second.get(); return r.first->second.get();
} catch (std::out_of_range& oor) { } catch (std::out_of_range& oor) {
LOG(error) << "Could not get remote region with id '" << id << "'. Does the region creator run with the same session id?"; LOG(error) << "Could not get remote region with id '" << id << "'. Does the region creator run with the same session id?";
@ -496,12 +508,14 @@ class Manager
fRegions.at(id)->StopAcks(); fRegions.at(id)->StopAcks();
{ {
boost::interprocess::scoped_lock<boost::interprocess::named_mutex> lock(fShmMtx); boost::interprocess::scoped_lock<boost::interprocess::named_mutex> lock(fShmMtx);
fShmRegions->at(id).fDestroyed = true; if (fRegions.at(id)->fRemoveOnDestruction) {
fShmRegions->at(id).fDestroyed = true;
(fEventCounter->fCount)++;
}
fRegions.erase(id); fRegions.erase(id);
(fEventCounter->fCount)++;
} }
fRegionEventsShmCV.notify_all(); fRegionEventsShmCV.notify_all();
} catch(std::out_of_range& oor) { } catch (std::out_of_range& oor) {
LOG(debug) << "RemoveRegion() could not locate region with id '" << id << "'"; LOG(debug) << "RemoveRegion() could not locate region with id '" << id << "'";
} }
fRegionsGen += 1; // signal TL cache invalidation fRegionsGen += 1; // signal TL cache invalidation
@ -749,7 +763,7 @@ class Manager
DecrementShmMsgCounter(segmentId); DecrementShmMsgCounter(segmentId);
try { try {
fMsgDebug->at(segmentId).erase(GetHandleFromAddress(ShmHeader::UserPtr(ptr), fSegmentId)); fMsgDebug->at(segmentId).erase(GetHandleFromAddress(ShmHeader::UserPtr(ptr), fSegmentId));
} catch(const std::out_of_range& oor) { } catch (const std::out_of_range& oor) {
LOG(debug) << "could not locate debug container for " << segmentId << ": " << oor.what(); LOG(debug) << "could not locate debug container for " << segmentId << ": " << oor.what();
} }
#endif #endif

4
fairmq/shmem/Message.h
View File

@ -381,9 +381,9 @@ class Message final : public fair::mq::Message
Deallocate(); Deallocate();
fAlignment = 0; fAlignment = 0;
fManager.DecrementMsgCounter(); fManager.DecrementMsgCounter();
} catch(SharedMemoryError& sme) { } catch (SharedMemoryError& sme) {
LOG(error) << "error closing message: " << sme.what(); LOG(error) << "error closing message: " << sme.what();
} catch(boost::interprocess::lock_exception& le) { } catch (boost::interprocess::lock_exception& le) {
LOG(error) << "error closing message: " << le.what(); LOG(error) << "error closing message: " << le.what();
} }
} }

48
fairmq/shmem/Monitor.cxx
View File

@ -8,6 +8,7 @@
#include "Monitor.h" #include "Monitor.h"
#include "Common.h" #include "Common.h"
#include "Region.h"
#include <fairmq/tools/IO.h> #include <fairmq/tools/IO.h>
#include <fairmq/tools/Strings.h> #include <fairmq/tools/Strings.h>
@ -179,15 +180,23 @@ bool Monitor::PrintShm(const ShmId& shmId)
managed_shared_memory managementSegment(open_read_only, std::string("fmq_" + shmId.shmId + "_mng").c_str()); managed_shared_memory managementSegment(open_read_only, std::string("fmq_" + shmId.shmId + "_mng").c_str());
VoidAlloc allocInstance(managementSegment.get_segment_manager()); VoidAlloc allocInstance(managementSegment.get_segment_manager());
Uint16SegmentInfoHashMap* segmentInfos = managementSegment.find<Uint16SegmentInfoHashMap>(unique_instance).first; Uint16SegmentInfoHashMap* shmSegments = managementSegment.find<Uint16SegmentInfoHashMap>(unique_instance).first;
std::unordered_map<uint16_t, boost::variant<RBTreeBestFitSegment, SimpleSeqFitSegment>> segments; std::unordered_map<uint16_t, boost::variant<RBTreeBestFitSegment, SimpleSeqFitSegment>> segments;
if (!segmentInfos) { Uint16RegionInfoHashMap* shmRegions = managementSegment.find<Uint16RegionInfoHashMap>(unique_instance).first;
std::unordered_map<uint16_t, std::unique_ptr<Region>> regions;
if (!shmSegments) {
LOG(error) << "Found management segment, but cannot locate segment info, something went wrong..."; LOG(error) << "Found management segment, but cannot locate segment info, something went wrong...";
return false; return false;
} }
for (const auto& s : *segmentInfos) { if (!shmSegments) {
LOG(error) << "Found management segment, but cannot locate region info...";
return false;
}
for (const auto& s : *shmSegments) {
if (s.second.fAllocationAlgorithm == AllocationAlgorithm::rbtree_best_fit) { 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, std::string("fmq_" + shmId.shmId + "_m_" + to_string(s.first)).c_str()));
} else { } else {
@ -221,7 +230,7 @@ bool Monitor::PrintShm(const ShmId& shmId)
<< ", session: " << sessionName << ", session: " << sessionName
<< ", creator id: " << creatorId << ", creator id: " << creatorId
<< ", devices: " << numDevices << ", devices: " << numDevices
<< ", segments:\n"; << ", managed segments:\n";
for (const auto& s : segments) { for (const auto& s : segments) {
size_t free = boost::apply_visitor(SegmentFreeMemory(), s.second); size_t free = boost::apply_visitor(SegmentFreeMemory(), s.second);
@ -243,6 +252,13 @@ bool Monitor::PrintShm(const ShmId& shmId)
<< "total: " << mtotal << "total: " << mtotal
<< ", free: " << mfree << ", free: " << mfree
<< ", used: " << mused; << ", used: " << mused;
if (!shmRegions->empty()) {
ss << "\n unmanaged regions:\n";
for (const auto& r : *shmRegions) {
ss << " [" << r.first << "]: " << (r.second.fDestroyed ? "destroyed" : "alive");
}
}
LOGV(info, user1) << ss.str(); LOGV(info, user1) << ss.str();
} catch (bie&) { } catch (bie&) {
return false; return false;
@ -458,15 +474,15 @@ unsigned long Monitor::GetFreeMemory(const ShmId& shmId, uint16_t segmentId)
boost::interprocess::named_mutex mtx(boost::interprocess::open_only, std::string("fmq_" + shmId.shmId + "_mtx").c_str()); boost::interprocess::named_mutex mtx(boost::interprocess::open_only, std::string("fmq_" + shmId.shmId + "_mtx").c_str());
boost::interprocess::scoped_lock<bipc::named_mutex> lock(mtx); boost::interprocess::scoped_lock<bipc::named_mutex> lock(mtx);
Uint16SegmentInfoHashMap* segmentInfos = managementSegment.find<Uint16SegmentInfoHashMap>(unique_instance).first; Uint16SegmentInfoHashMap* shmSegments = managementSegment.find<Uint16SegmentInfoHashMap>(unique_instance).first;
if (!segmentInfos) { if (!shmSegments) {
LOG(error) << "Found management segment, but could not locate segment info"; LOG(error) << "Found management segment, but could not locate segment info";
throw MonitorError("Found management segment, but could not locate segment info"); throw MonitorError("Found management segment, but could not locate segment info");
} }
auto it = segmentInfos->find(segmentId); auto it = shmSegments->find(segmentId);
if (it != segmentInfos->end()) { if (it != shmSegments->end()) {
if (it->second.fAllocationAlgorithm == AllocationAlgorithm::rbtree_best_fit) { 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, std::string("fmq_" + shmId.shmId + "_m_" + std::to_string(segmentId)).c_str());
return segment.get_free_memory(); return segment.get_free_memory();
@ -531,12 +547,12 @@ std::vector<std::pair<std::string, bool>> Monitor::Cleanup(const ShmId& shmId, b
try { try {
bipc::managed_shared_memory managementSegment(bipc::open_only, managementSegmentName.c_str()); bipc::managed_shared_memory managementSegment(bipc::open_only, managementSegmentName.c_str());
Uint16RegionInfoHashMap* regions = managementSegment.find<Uint16RegionInfoHashMap>(bipc::unique_instance).first; Uint16RegionInfoHashMap* shmRegions = managementSegment.find<Uint16RegionInfoHashMap>(bipc::unique_instance).first;
if (regions) { if (shmRegions) {
if (verbose) { if (verbose) {
LOG(info) << "Found " << regions->size() << " unmanaged regions..."; LOG(info) << "Found " << shmRegions->size() << " unmanaged regions...";
} }
for (const auto& region : *regions) { for (const auto& region : *shmRegions) {
uint16_t id = region.first; uint16_t id = region.first;
RegionInfo info = region.second; RegionInfo info = region.second;
string path = info.fPath.c_str(); string path = info.fPath.c_str();
@ -553,13 +569,13 @@ std::vector<std::pair<std::string, bool>> Monitor::Cleanup(const ShmId& shmId, b
} }
} }
Uint16SegmentInfoHashMap* segments = managementSegment.find<Uint16SegmentInfoHashMap>(bipc::unique_instance).first; Uint16SegmentInfoHashMap* shmSegments = managementSegment.find<Uint16SegmentInfoHashMap>(bipc::unique_instance).first;
if (segments) { if (shmSegments) {
if (verbose) { if (verbose) {
LOG(info) << "Found " << segments->size() << " managed segments..."; LOG(info) << "Found " << shmSegments->size() << " managed segments...";
} }
for (const auto& segment : *segments) { for (const auto& segment : *shmSegments) {
result.emplace_back(Remove<bipc::shared_memory_object>("fmq_" + shmId.shmId + "_m_" + to_string(segment.first), verbose)); result.emplace_back(Remove<bipc::shared_memory_object>("fmq_" + shmId.shmId + "_m_" + to_string(segment.first), verbose));
} }
} else { } else {

78
fairmq/shmem/Region.h
View File

@ -38,12 +38,13 @@ namespace fair::mq::shmem
struct Region struct Region
{ {
Region(const std::string& shmId, uint16_t id, uint64_t size, bool remote, RegionCallback callback, RegionBulkCallback bulkCallback, RegionConfig cfg) Region(const std::string& shmId, uint64_t size, bool remote, RegionCallback callback, RegionBulkCallback bulkCallback, RegionConfig cfg)
: fRemote(remote) : fRemote(remote)
, fRemoveOnDestruction(cfg.removeOnDestruction)
, fLinger(cfg.linger) , fLinger(cfg.linger)
, fStopAcks(false) , fStopAcks(false)
, fName("fmq_" + shmId + "_rg_" + std::to_string(id)) , fName("fmq_" + shmId + "_rg_" + std::to_string(cfg.id.value()))
, fQueueName("fmq_" + shmId + "_rgq_" + std::to_string(id)) , fQueueName("fmq_" + shmId + "_rgq_" + std::to_string(cfg.id.value()))
, fShmemObject() , fShmemObject()
, fFile(nullptr) , fFile(nullptr)
, fFileMapping() , fFileMapping()
@ -81,23 +82,46 @@ struct Region
if (fRemote) { if (fRemote) {
fShmemObject = shared_memory_object(open_only, fName.c_str(), read_write); fShmemObject = shared_memory_object(open_only, fName.c_str(), read_write);
} else { } else {
fShmemObject = shared_memory_object(create_only, fName.c_str(), read_write); switch (cfg.constructionMode) {
fShmemObject.truncate(size); case RegionConstruction::create:
fShmemObject = shared_memory_object(create_only, fName.c_str(), read_write);
fShmemObject.truncate(size);
break;
case RegionConstruction::open:
fShmemObject = shared_memory_object(open_only, fName.c_str(), read_write);
break;
case RegionConstruction::open_or_create:
try {
fShmemObject = shared_memory_object(open_only, fName.c_str(), read_write);
} catch (interprocess_exception&) {
fShmemObject = shared_memory_object(create_only, fName.c_str(), read_write);
fShmemObject.truncate(size);
}
break;
default:
throw TransportError(tools::ToString("Unknown RegionConstruction mode provided "));
break;
}
} }
} catch (interprocess_exception& e) { } catch (interprocess_exception& e) {
LOG(error) << "Failed " << (fRemote ? "opening" : "creating") << " shared_memory_object for region id '" << id << "': " << e.what(); LOG(error) << "Failed " << (fRemote ? "opening" : "creating") << " shared_memory_object for region id '" << cfg.id.value() << "': " << e.what();
throw; throw;
} }
try { try {
fRegion = mapped_region(fShmemObject, read_write, 0, 0, 0, cfg.creationFlags); fRegion = mapped_region(fShmemObject, read_write, 0, 0, 0, cfg.creationFlags);
} catch (interprocess_exception& e) { } catch (interprocess_exception& e) {
LOG(error) << "Failed mapping shared_memory_object for region id '" << id << "': " << e.what(); LOG(error) << "Failed mapping shared_memory_object for region id '" << cfg.id.value() << "': " << e.what();
throw; throw;
} }
} }
InitializeQueues(); // skip queues initialization if region is being created without callbacks passed
StartSendingAcks(); if (fRemote || (fCallback || fBulkCallback)) {
InitializeQueues();
StartAckSender();
} else {
LOG(trace) << "skipping queues creation & ack sender thread, because created region locally but no callback were provided";
}
LOG(trace) << "shmem: initialized region: " << fName << " (" << (fRemote ? "remote" : "local") << ")"; LOG(trace) << "shmem: initialized region: " << fName << " (" << (fRemote ? "remote" : "local") << ")";
} }
@ -112,16 +136,11 @@ struct Region
void InitializeQueues() void InitializeQueues()
{ {
using namespace boost::interprocess; using namespace boost::interprocess;
fQueue = std::make_unique<message_queue>(open_or_create, fQueueName.c_str(), 1024, fAckBunchSize * sizeof(RegionBlock));
if (fRemote) {
fQueue = std::make_unique<message_queue>(open_only, fQueueName.c_str());
} else {
fQueue = std::make_unique<message_queue>(create_only, fQueueName.c_str(), 1024, fAckBunchSize * sizeof(RegionBlock));
}
LOG(trace) << "shmem: initialized region queue: " << fQueueName << " (" << (fRemote ? "remote" : "local") << ")"; LOG(trace) << "shmem: initialized region queue: " << fQueueName << " (" << (fRemote ? "remote" : "local") << ")";
} }
void StartSendingAcks() void StartAckSender()
{ {
fAcksSender = std::thread(&Region::SendAcks, this); fAcksSender = std::thread(&Region::SendAcks, this);
} }
@ -163,7 +182,7 @@ struct Region
<< " blocks left to send: " << blocksToSend << ")."; << " blocks left to send: " << blocksToSend << ").";
} }
void StartReceivingAcks() void StartAckReceiver()
{ {
if (!fAcksReceiver.joinable()) { if (!fAcksReceiver.joinable()) {
fAcksReceiver = std::thread(&Region::ReceiveAcks, this); fAcksReceiver = std::thread(&Region::ReceiveAcks, this);
@ -255,20 +274,26 @@ struct Region
fAcksReceiver.join(); fAcksReceiver.join();
} }
if (boost::interprocess::shared_memory_object::remove(fName.c_str())) { if (fRemoveOnDestruction) {
LOG(trace) << "Region '" << fName << "' destroyed."; if (boost::interprocess::shared_memory_object::remove(fName.c_str())) {
} LOG(trace) << "Region '" << fName << "' destroyed.";
}
if (boost::interprocess::file_mapping::remove(fName.c_str())) { if (boost::interprocess::file_mapping::remove(fName.c_str())) {
LOG(trace) << "File mapping '" << fName << "' destroyed."; LOG(trace) << "File mapping '" << fName << "' destroyed.";
} }
} else {
if (fFile) { LOG(debug) << "Skipping removal of " << fName << " unmanaged region, because RegionConfig::removeOnDestruction is false";
fclose(fFile);
} }
if (boost::interprocess::message_queue::remove(fQueueName.c_str())) { if (boost::interprocess::message_queue::remove(fQueueName.c_str())) {
LOG(trace) << "Region queue '" << fQueueName << "' destroyed."; LOG(trace) << "Region queue '" << fQueueName << "' destroyed.";
} else {
LOG(debug) << "Region queue '" << fQueueName << "' not destroyed.";
}
if (fFile) {
fclose(fFile);
} }
} else { } else {
// LOG(debug) << "Region queue '" << fQueueName << "' is remote, no cleanup necessary"; // LOG(debug) << "Region queue '" << fQueueName << "' is remote, no cleanup necessary";
@ -278,6 +303,7 @@ struct Region
} }
bool fRemote; bool fRemote;
bool fRemoveOnDestruction;
uint32_t fLinger; uint32_t fLinger;
std::atomic<bool> fStopAcks; std::atomic<bool> fStopAcks;
std::string fName; std::string fName;

12
fairmq/shmem/TransportFactory.h
View File

@ -147,14 +147,14 @@ class TransportFactory final : public fair::mq::TransportFactory
{ {
cfg.path = path; cfg.path = path;
cfg.creationFlags = flags; cfg.creationFlags = flags;
return CreateUnmanagedRegion(size, callback, nullptr, cfg); return CreateUnmanagedRegion(size, callback, nullptr, std::move(cfg));
} }
UnmanagedRegionPtr CreateUnmanagedRegion(size_t size, RegionBulkCallback bulkCallback = nullptr, const std::string& path = "", int flags = 0, fair::mq::RegionConfig cfg = fair::mq::RegionConfig()) override UnmanagedRegionPtr CreateUnmanagedRegion(size_t size, RegionBulkCallback bulkCallback = nullptr, const std::string& path = "", int flags = 0, fair::mq::RegionConfig cfg = fair::mq::RegionConfig()) override
{ {
cfg.path = path; cfg.path = path;
cfg.creationFlags = flags; cfg.creationFlags = flags;
return CreateUnmanagedRegion(size, nullptr, bulkCallback, cfg); return CreateUnmanagedRegion(size, nullptr, bulkCallback, std::move(cfg));
} }
UnmanagedRegionPtr CreateUnmanagedRegion(size_t size, int64_t userFlags, RegionCallback callback = nullptr, const std::string& path = "", int flags = 0, fair::mq::RegionConfig cfg = fair::mq::RegionConfig()) override UnmanagedRegionPtr CreateUnmanagedRegion(size_t size, int64_t userFlags, RegionCallback callback = nullptr, const std::string& path = "", int flags = 0, fair::mq::RegionConfig cfg = fair::mq::RegionConfig()) override
@ -162,7 +162,7 @@ class TransportFactory final : public fair::mq::TransportFactory
cfg.path = path; cfg.path = path;
cfg.userFlags = userFlags; cfg.userFlags = userFlags;
cfg.creationFlags = flags; cfg.creationFlags = flags;
return CreateUnmanagedRegion(size, callback, nullptr, cfg); return CreateUnmanagedRegion(size, callback, nullptr, std::move(cfg));
} }
UnmanagedRegionPtr CreateUnmanagedRegion(size_t size, int64_t userFlags, RegionBulkCallback bulkCallback = nullptr, const std::string& path = "", int flags = 0, fair::mq::RegionConfig cfg = fair::mq::RegionConfig()) override UnmanagedRegionPtr CreateUnmanagedRegion(size_t size, int64_t userFlags, RegionBulkCallback bulkCallback = nullptr, const std::string& path = "", int flags = 0, fair::mq::RegionConfig cfg = fair::mq::RegionConfig()) override
@ -170,16 +170,16 @@ class TransportFactory final : public fair::mq::TransportFactory
cfg.path = path; cfg.path = path;
cfg.userFlags = userFlags; cfg.userFlags = userFlags;
cfg.creationFlags = flags; cfg.creationFlags = flags;
return CreateUnmanagedRegion(size, nullptr, bulkCallback, cfg); return CreateUnmanagedRegion(size, nullptr, bulkCallback, std::move(cfg));
} }
UnmanagedRegionPtr CreateUnmanagedRegion(size_t size, RegionCallback callback, RegionConfig cfg) override UnmanagedRegionPtr CreateUnmanagedRegion(size_t size, RegionCallback callback, RegionConfig cfg) override
{ {
return CreateUnmanagedRegion(size, callback, nullptr, cfg); return CreateUnmanagedRegion(size, callback, nullptr, std::move(cfg));
} }
UnmanagedRegionPtr CreateUnmanagedRegion(size_t size, RegionBulkCallback bulkCallback, RegionConfig cfg) override UnmanagedRegionPtr CreateUnmanagedRegion(size_t size, RegionBulkCallback bulkCallback, RegionConfig cfg) override
{ {
return CreateUnmanagedRegion(size, nullptr, bulkCallback, cfg); return CreateUnmanagedRegion(size, nullptr, bulkCallback, std::move(cfg));
} }
UnmanagedRegionPtr CreateUnmanagedRegion(size_t size, RegionCallback callback, RegionBulkCallback bulkCallback, fair::mq::RegionConfig cfg) UnmanagedRegionPtr CreateUnmanagedRegion(size_t size, RegionCallback callback, RegionBulkCallback bulkCallback, fair::mq::RegionConfig cfg)