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

Add region events subscriptions

Browse Source
This commit is contained in:
Alexey Rybalchenko
2020-04-26 22:41:22 +02:00
parent 5721ea9510
commit e1a113aabe
35 changed files with 445 additions and 188 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
fairmq/shmem/Common.h
View File

@@ -41,15 +41,21 @@ struct RegionInfo
RegionInfo(const VoidAlloc& alloc)
: fPath("", alloc)
, fFlags(0)
, fUserFlags(0)
, fDestroyed(false)
{}
RegionInfo(const char* path, int flags, const VoidAlloc& alloc)
RegionInfo(const char* path, const int flags, const uint64_t userFlags, const VoidAlloc& alloc)
: fPath(path, alloc)
, fFlags(flags)
, fUserFlags(userFlags)
, fDestroyed(false)
{}
Str fPath;
int fFlags;
uint64_t fUserFlags;
bool fDestroyed;
};
using Uint64RegionInfoPairAlloc = boost::interprocess::allocator<std::pair<const uint64_t, RegionInfo>, SegmentManager>;

219
fairmq/shmem/Manager.cxx
View File

@@ -7,7 +7,6 @@
********************************************************************************/
#include "Manager.h"
#include "Common.h"
#include <fairmq/tools/CppSTL.h>
#include <fairmq/tools/Strings.h>
@@ -19,6 +18,7 @@ using namespace std;
using bie = ::boost::interprocess::interprocess_exception;
namespace bipc = ::boost::interprocess;
namespace bfs = ::boost::filesystem;
namespace bpt = ::boost::posix_time;
namespace fair
{
@@ -27,19 +27,26 @@ namespace mq
namespace shmem
{
std::unordered_map<uint64_t, std::unique_ptr<Region>> Manager::fRegions;
Manager::Manager(const std::string& id, size_t size)
Manager::Manager(const string& id, size_t size)
: fShmId(id)
, fSegmentName("fmq_" + fShmId + "_main")
, fManagementSegmentName("fmq_" + fShmId + "_mng")
, fSegment(bipc::open_or_create, fSegmentName.c_str(), size)
, fManagementSegment(bipc::open_or_create, fManagementSegmentName.c_str(), 65536)
, fManagementSegment(bipc::open_or_create, fManagementSegmentName.c_str(), 655360)
, fShmVoidAlloc(fManagementSegment.get_segment_manager())
, fShmMtx(bipc::open_or_create, string("fmq_" + fShmId + "_mtx").c_str())
, fRegionEventsCV(bipc::open_or_create, string("fmq_" + fShmId + "_cv").c_str())
, fRegionEventsSubscriptionActive(false)
, fDeviceCounter(nullptr)
, fRegionInfos(nullptr)
, fInterrupted(false)
{
LOG(debug) << "created/opened shared memory segment '" << "fmq_" << fShmId << "_main" << "' of " << size << " bytes. Available are " << fSegment.get_free_memory() << " bytes.";
fRegionInfos = fManagementSegment.find_or_construct<Uint64RegionInfoMap>(bipc::unique_instance)(fShmVoidAlloc);
// store info about the managed segment as region with id 0
fRegionInfos->emplace(0, RegionInfo("", 0, 0, fShmVoidAlloc));
bipc::scoped_lock<bipc::named_mutex> lock(fShmMtx);
fDeviceCounter = fManagementSegment.find<DeviceCounter>(bipc::unique_instance).first;
@@ -55,7 +62,7 @@ Manager::Manager(const std::string& id, size_t size)
}
}
void Manager::StartMonitor(const std::string& id)
void Manager::StartMonitor(const string& id)
{
try {
bipc::named_mutex monitorStatus(bipc::open_only, string("fmq_" + id + "_ms").c_str());
@@ -94,47 +101,74 @@ void Manager::StartMonitor(const std::string& id)
}
}
void Manager::Interrupt()
pair<bipc::mapped_region*, uint64_t> Manager::CreateRegion(const size_t size, const int64_t userFlags, RegionCallback callback, const string& path /* = "" */, int flags /* = 0 */)
{
}
try {
void Manager::Resume()
{
// close remote regions before processing new transfers
for (auto it = fRegions.begin(); it != fRegions.end(); /**/) {
if (it->second->fRemote) {
it = fRegions.erase(it);
} else {
++it;
}
}
}
pair<bipc::mapped_region*, uint64_t> result;
bipc::mapped_region* Manager::CreateRegion(const size_t size, const uint64_t id, RegionCallback callback, const std::string& path /* = "" */, int flags /* = 0 */)
{
auto it = fRegions.find(id);
if (it != fRegions.end()) {
LOG(error) << "Trying to create a region that already exists";
return nullptr;
} else {
// create region info
{
uint64_t id = 0;
bipc::scoped_lock<bipc::named_mutex> lock(fShmMtx);
VoidAlloc voidAlloc(fManagementSegment.get_segment_manager());
Uint64RegionInfoMap* infoMap = fManagementSegment.find_or_construct<Uint64RegionInfoMap>(bipc::unique_instance)(voidAlloc);
infoMap->emplace(id, RegionInfo(path.c_str(), flags, voidAlloc));
RegionCounter* rc = fManagementSegment.find<RegionCounter>(bipc::unique_instance).first;
if (rc) {
LOG(debug) << "region counter found, with value of " << rc->fCount << ". incrementing.";
(rc->fCount)++;
LOG(debug) << "incremented region counter, now: " << rc->fCount;
} else {
LOG(debug) << "no region counter found, creating one and initializing with 1";
rc = fManagementSegment.construct<RegionCounter>(bipc::unique_instance)(1);
LOG(debug) << "initialized region counter with: " << rc->fCount;
}
id = rc->fCount;
auto it = fRegions.find(id);
if (it != fRegions.end()) {
LOG(error) << "Trying to create a region that already exists";
return {nullptr, id};
}
// create region info
fRegionInfos->emplace(id, RegionInfo(path.c_str(), flags, userFlags, fShmVoidAlloc));
auto r = fRegions.emplace(id, tools::make_unique<Region>(*this, id, size, false, callback, path, flags));
// LOG(debug) << "Created region with id '" << id << "', path: '" << path << "', flags: '" << flags << "'";
r.first->second->StartReceivingAcks();
result.first = &(r.first->second->fRegion);
result.second = id;
}
// LOG(debug) << "Created region with id '" << id << "', path: '" << path << "', flags: '" << flags << "'";
fRegionEventsCV.notify_all();
auto r = fRegions.emplace(id, tools::make_unique<Region>(*this, id, size, false, callback, path, flags));
return result;
r.first->second->StartReceivingAcks();
return &(r.first->second->fRegion);
} catch (bipc::interprocess_exception& e) {
LOG(error) << "cannot create region. Already created/not cleaned up?";
LOG(error) << e.what();
throw;
}
}
Region* Manager::GetRemoteRegion(const uint64_t id)
void Manager::RemoveRegion(const uint64_t id)
{
{
bipc::scoped_lock<bipc::named_mutex> lock(fShmMtx);
fRegions.erase(id);
fRegionInfos->at(id).fDestroyed = true;
}
fRegionEventsCV.notify_all();
}
Region* Manager::GetRegion(const uint64_t id)
{
bipc::scoped_lock<bipc::named_mutex> lock(fShmMtx);
return GetRegionUnsafe(id);
}
Region* Manager::GetRegionUnsafe(const uint64_t id)
{
// remote region could actually be a local one if a message originates from this device (has been sent out and returned)
auto it = fRegions.find(id);
@@ -142,36 +176,108 @@ Region* Manager::GetRemoteRegion(const uint64_t id)
return it->second.get();
} else {
try {
string path;
int flags;
// get region info
{
bipc::scoped_lock<bipc::named_mutex> lock(fShmMtx);
Uint64RegionInfoMap* infoMap = fManagementSegment.find<Uint64RegionInfoMap>(bipc::unique_instance).first;
if (infoMap == nullptr) {
LOG(error) << "Unable to locate the region info";
throw SharedMemoryError("Unable to locate remote region info");
}
RegionInfo regionInfo = infoMap->at(id);
path = regionInfo.fPath.c_str();
flags = regionInfo.fFlags;
}
RegionInfo regionInfo = fRegionInfos->at(id);
string path = regionInfo.fPath.c_str();
int flags = regionInfo.fFlags;
// LOG(debug) << "Located remote region with id '" << id << "', path: '" << path << "', flags: '" << flags << "'";
auto r = fRegions.emplace(id, tools::make_unique<Region>(*this, id, 0, true, nullptr, path, flags));
r.first->second->StartSendingAcks();
return r.first->second.get();
} catch (bie& e) {
LOG(warn) << "Could not get remote region for id: " << id;
return nullptr;
}
}
}
void Manager::RemoveRegion(const uint64_t id)
vector<fair::mq::RegionInfo> Manager::GetRegionInfo()
{
fRegions.erase(id);
bipc::scoped_lock<bipc::named_mutex> lock(fShmMtx);
return GetRegionInfoUnsafe();
}
vector<fair::mq::RegionInfo> Manager::GetRegionInfoUnsafe()
{
vector<fair::mq::RegionInfo> result;
for (const auto& e : *fRegionInfos) {
fair::mq::RegionInfo info;
info.id = e.first;
info.flags = e.second.fUserFlags;
info.event = e.second.fDestroyed ? RegionEvent::destroyed : RegionEvent::created;
if (info.id != 0) {
if (!e.second.fDestroyed) {
auto region = GetRegionUnsafe(info.id);
info.ptr = region->fRegion.get_address();
info.size = region->fRegion.get_size();
} else {
info.ptr = nullptr;
info.size = 0;
}
result.push_back(info);
} else {
if (!e.second.fDestroyed) {
info.ptr = fSegment.get_address();
info.size = fSegment.get_size();
} else {
info.ptr = nullptr;
info.size = 0;
}
result.push_back(info);
}
}
return result;
}
void Manager::SubscribeToRegionEvents(RegionEventCallback callback)
{
bipc::scoped_lock<bipc::named_mutex> lock(fShmMtx);
if (fRegionEventThread.joinable()) {
fRegionEventsSubscriptionActive.store(false);
fRegionEventThread.join();
}
fRegionEventCallback = callback;
fRegionEventsSubscriptionActive.store(true);
fRegionEventThread = thread(&Manager::RegionEventsSubscription, this);
}
void Manager::UnsubscribeFromRegionEvents()
{
if (fRegionEventThread.joinable()) {
fRegionEventsSubscriptionActive.store(false);
fRegionEventsCV.notify_all();
fRegionEventThread.join();
}
bipc::scoped_lock<bipc::named_mutex> lock(fShmMtx);
fRegionEventCallback = nullptr;
}
void Manager::RegionEventsSubscription()
{
while (fRegionEventsSubscriptionActive.load()) {
bipc::scoped_lock<bipc::named_mutex> lock(fShmMtx);
auto infos = GetRegionInfoUnsafe();
for (const auto& i : infos) {
auto el = fObservedRegionEvents.find(i.id);
if (el == fObservedRegionEvents.end()) {
fRegionEventCallback(i);
fObservedRegionEvents.emplace(i.id, i.event);
} else {
if (el->second == RegionEvent::created && i.event == RegionEvent::destroyed) {
fRegionEventCallback(i);
el->second = i.event;
} else {
// LOG(debug) << "ignoring event for id" << i.id << ":";
// LOG(debug) << "incoming event: " << i.event;
// LOG(debug) << "stored event: " << el->second;
}
}
}
fRegionEventsCV.wait(lock);
}
}
void Manager::RemoveSegments()
@@ -193,6 +299,12 @@ Manager::~Manager()
{
bool lastRemoved = false;
if (fRegionEventThread.joinable()) {
fRegionEventsSubscriptionActive.store(false);
fRegionEventsCV.notify_all();
fRegionEventThread.join();
}
try {
bipc::scoped_lock<bipc::named_mutex> lock(fShmMtx);
@@ -212,6 +324,7 @@ Manager::~Manager()
if (lastRemoved) {
bipc::named_mutex::remove(string("fmq_" + fShmId + "_mtx").c_str());
bipc::named_condition::remove(string("fmq_" + fShmId + "_cv").c_str());
}
}

43
fairmq/shmem/Manager.h
View File

@@ -15,18 +15,24 @@
#ifndef FAIR_MQ_SHMEM_MANAGER_H_
#define FAIR_MQ_SHMEM_MANAGER_H_
#include "Region.h"
#include "Common.h"
#include "Region.h"
#include <FairMQLogger.h>
#include <FairMQUnmanagedRegion.h>
#include <boost/interprocess/managed_shared_memory.hpp>
#include <boost/interprocess/ipc/message_queue.hpp>
#include <boost/interprocess/managed_shared_memory.hpp>
#include <boost/interprocess/sync/named_condition.hpp>
#include <boost/interprocess/sync/named_mutex.hpp>
#include <string>
#include <unordered_map>
#include <set>
#include <stdexcept>
#include <string>
#include <thread>
#include <unordered_map>
#include <utility>
#include <vector>
namespace fair
{
@@ -56,17 +62,25 @@ class Manager
static void StartMonitor(const std::string&);
static void Interrupt();
static void Resume();
void Interrupt() { fInterrupted.store(true); }
void Resume() { fInterrupted.store(false); }
bool Interrupted() { return fInterrupted.load(); }
int GetDeviceCounter();
int IncrementDeviceCounter();
int DecrementDeviceCounter();
boost::interprocess::mapped_region* CreateRegion(const size_t size, const uint64_t id, RegionCallback callback, const std::string& path = "", int flags = 0);
Region* GetRemoteRegion(const uint64_t id);
std::pair<boost::interprocess::mapped_region*, uint64_t> CreateRegion(const size_t size, const int64_t userFlags, RegionCallback callback, const std::string& path = "", int flags = 0);
Region* GetRegion(const uint64_t id);
Region* GetRegionUnsafe(const uint64_t id);
void RemoveRegion(const uint64_t id);
std::vector<fair::mq::RegionInfo> GetRegionInfo();
std::vector<fair::mq::RegionInfo> GetRegionInfoUnsafe();
void SubscribeToRegionEvents(RegionEventCallback callback);
void UnsubscribeFromRegionEvents();
void RegionEventsSubscription();
void RemoveSegments();
private:
@@ -75,9 +89,20 @@ class Manager
std::string fManagementSegmentName;
boost::interprocess::managed_shared_memory fSegment;
boost::interprocess::managed_shared_memory fManagementSegment;
VoidAlloc fShmVoidAlloc;
boost::interprocess::named_mutex fShmMtx;
boost::interprocess::named_condition fRegionEventsCV;
std::thread fRegionEventThread;
std::atomic<bool> fRegionEventsSubscriptionActive;
std::function<void(fair::mq::RegionInfo)> fRegionEventCallback;
std::unordered_map<uint64_t, RegionEvent> fObservedRegionEvents;
DeviceCounter* fDeviceCounter;
static std::unordered_map<uint64_t, std::unique_ptr<Region>> fRegions;
Uint64RegionInfoMap* fRegionInfos;
std::unordered_map<uint64_t, std::unique_ptr<Region>> fRegions;
std::atomic<bool> fInterrupted;
};
} // namespace shmem

7
fairmq/shmem/Message.cxx
View File

@@ -29,7 +29,6 @@ namespace mq
namespace shmem
{
atomic<bool> Message::fInterrupted(false);
Transport Message::fTransportType = Transport::SHM;
Message::Message(Manager& manager, FairMQTransportFactory* factory)
@@ -113,7 +112,7 @@ bool Message::InitializeChunk(const size_t size)
} catch (bipc::bad_alloc& ba) {
// LOG(warn) << "Shared memory full...";
this_thread::sleep_for(chrono::milliseconds(50));
if (fInterrupted) {
if (fManager.Interrupted()) {
return false;
} else {
continue;
@@ -164,7 +163,7 @@ void* Message::GetData() const
fLocalPtr = nullptr;
}
} else {
fRegionPtr = fManager.GetRemoteRegion(fMeta.fRegionId);
fRegionPtr = fManager.GetRegion(fMeta.fRegionId);
if (fRegionPtr) {
fLocalPtr = reinterpret_cast<char*>(fRegionPtr->fRegion.get_address()) + fMeta.fHandle;
} else {
@@ -221,7 +220,7 @@ void Message::CloseMessage()
fMeta.fHandle = -1;
} else {
if (!fRegionPtr) {
fRegionPtr = fManager.GetRemoteRegion(fMeta.fRegionId);
fRegionPtr = fManager.GetRegion(fMeta.fRegionId);
}
if (fRegionPtr) {

1
fairmq/shmem/Message.h
View File

@@ -65,7 +65,6 @@ class Message final : public fair::mq::Message
mutable Region* fRegionPtr;
mutable char* fLocalPtr;
static std::atomic<bool> fInterrupted;
static Transport fTransportType;
bool InitializeChunk(const size_t size);

11
fairmq/shmem/Monitor.cxx
View File

@@ -15,6 +15,7 @@
#include <boost/interprocess/file_mapping.hpp>
#include <boost/interprocess/sync/named_mutex.hpp>
#include <boost/interprocess/sync/named_condition.hpp>
#include <boost/interprocess/ipc/message_queue.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
@@ -427,6 +428,15 @@ void Monitor::RemoveMutex(const string& name)
}
}
void Monitor::RemoveCondition(const string& name)
{
if (bipc::named_condition::remove(name.c_str())) {
cout << "Successfully removed \"" << name << "\"." << endl;
} else {
cout << "Did not remove \"" << name << "\". Already removed?" << endl;
}
}
void Monitor::Cleanup(const string& shmId)
{
string managementSegmentName("fmq_" + shmId + "_mng");
@@ -469,6 +479,7 @@ void Monitor::Cleanup(const string& shmId)
RemoveObject("fmq_" + shmId + "_main");
RemoveMutex("fmq_" + shmId + "_mtx");
RemoveCondition("fmq_" + shmId + "_cv");
cout << endl;
}

1
fairmq/shmem/Monitor.h
View File

@@ -42,6 +42,7 @@ class Monitor
static void RemoveFileMapping(const std::string&);
static void RemoveQueue(const std::string&);
static void RemoveMutex(const std::string&);
static void RemoveCondition(const std::string&);
struct DaemonPresent : std::runtime_error { using std::runtime_error::runtime_error; };

13
fairmq/shmem/Region.cxx
View File

@@ -71,7 +71,6 @@ Region::Region(Manager& manager, uint64_t id, uint64_t size, bool remote, Region
InitializeQueues();
LOG(debug) << "shmem: initialized region: " << fName;
fSendAcksWorker = thread(&Region::SendAcks, this);
}
void Region::InitializeQueues()
@@ -84,6 +83,11 @@ void Region::InitializeQueues()
LOG(debug) << "shmem: initialized region queue: " << fQueueName;
}
void Region::StartSendingAcks()
{
fSendAcksWorker = thread(&Region::SendAcks, this);
}
void Region::StartReceivingAcks()
{
fReceiveAcksWorker = thread(&Region::ReceiveAcks, this);
@@ -114,12 +118,12 @@ void Region::ReceiveAcks()
void Region::ReleaseBlock(const RegionBlock &block)
{
unique_lock<mutex> lock(fBlockLock);
unique_lock<mutex> lock(fBlockMtx);
fBlocksToFree.emplace_back(block);
if (fBlocksToFree.size() >= fAckBunchSize) {
lock.unlock(); // reduces contention on fBlockLock
lock.unlock(); // reduces contention on fBlockMtx
fBlockSendCV.notify_one();
}
}
@@ -132,7 +136,7 @@ void Region::SendAcks()
size_t blocksToSend = 0;
{ // mutex locking block
unique_lock<mutex> lock(fBlockLock);
unique_lock<mutex> lock(fBlockMtx);
// try to get more blocks without waiting (we can miss a notify from CloseMessage())
if (!fStop && (fBlocksToFree.size() < fAckBunchSize)) {
@@ -166,6 +170,7 @@ Region::~Region()
fStop = true;
if (fSendAcksWorker.joinable()) {
fBlockSendCV.notify_one();
fSendAcksWorker.join();
}

6
fairmq/shmem/Region.h
View File

@@ -49,11 +49,11 @@ struct Region
void InitializeQueues();
void StartSendingAcks();
void SendAcks();
void StartReceivingAcks();
void ReceiveAcks();
void ReleaseBlock(const RegionBlock &);
void SendAcks();
~Region();
@@ -67,7 +67,7 @@ struct Region
boost::interprocess::file_mapping fFileMapping;
boost::interprocess::mapped_region fRegion;
std::mutex fBlockLock;
std::mutex fBlockMtx;
std::condition_variable fBlockSendCV;
std::vector<RegionBlock> fBlocksToFree;
const std::size_t fAckBunchSize = 256;

30
fairmq/shmem/Socket.cxx
View File

@@ -28,8 +28,6 @@ namespace mq
namespace shmem
{
atomic<bool> Socket::fInterrupted(false);
struct ZMsg
{
ZMsg() { int rc __attribute__((unused)) = zmq_msg_init(&fMsg); assert(rc == 0); }
@@ -133,7 +131,7 @@ int Socket::Send(MessagePtr& msg, const int timeout)
ZMsg zmqMsg(sizeof(MetaHeader));
std::memcpy(zmqMsg.Data(), &(shmMsg->fMeta), sizeof(MetaHeader));
while (true && !fInterrupted) {
while (true && !fManager.Interrupted()) {
int nbytes = zmq_msg_send(zmqMsg.Msg(), fSocket, flags);
if (nbytes > 0) {
shmMsg->fQueued = true;
@@ -142,7 +140,7 @@ int Socket::Send(MessagePtr& msg, const int timeout)
fBytesTx += size;
return size;
} else if (zmq_errno() == EAGAIN) {
if (!fInterrupted && ((flags & ZMQ_DONTWAIT) == 0)) {
if (!fManager.Interrupted() && ((flags & ZMQ_DONTWAIT) == 0)) {
if (timeout > 0) {
elapsed += fSndTimeout;
if (elapsed >= timeout) {
@@ -198,7 +196,7 @@ int Socket::Receive(MessagePtr& msg, const int timeout)
++fMessagesRx;
return size;
} else if (zmq_errno() == EAGAIN) {
if (!fInterrupted && ((flags & ZMQ_DONTWAIT) == 0)) {
if (!fManager.Interrupted() && ((flags & ZMQ_DONTWAIT) == 0)) {
if (timeout > 0) {
elapsed += fRcvTimeout;
if (elapsed >= timeout) {
@@ -242,7 +240,7 @@ int64_t Socket::Send(vector<MessagePtr>& msgVec, const int timeout)
std::memcpy(metas++, &(shmMsg->fMeta), sizeof(MetaHeader));
}
while (!fInterrupted) {
while (!fManager.Interrupted()) {
int64_t totalSize = 0;
int nbytes = zmq_msg_send(zmqMsg.Msg(), fSocket, flags);
if (nbytes > 0) {
@@ -260,7 +258,7 @@ int64_t Socket::Send(vector<MessagePtr>& msgVec, const int timeout)
return totalSize;
} else if (zmq_errno() == EAGAIN) {
if (!fInterrupted && ((flags & ZMQ_DONTWAIT) == 0)) {
if (!fManager.Interrupted() && ((flags & ZMQ_DONTWAIT) == 0)) {
if (timeout > 0) {
elapsed += fSndTimeout;
if (elapsed >= timeout) {
@@ -296,7 +294,7 @@ int64_t Socket::Receive(vector<MessagePtr>& msgVec, const int timeout)
ZMsg zmqMsg;
while (!fInterrupted) {
while (!fManager.Interrupted()) {
int64_t totalSize = 0;
int nbytes = zmq_msg_recv(zmqMsg.Msg(), fSocket, flags);
if (nbytes > 0) {
@@ -327,7 +325,7 @@ int64_t Socket::Receive(vector<MessagePtr>& msgVec, const int timeout)
return totalSize;
} else if (zmq_errno() == EAGAIN) {
if (!fInterrupted && ((flags & ZMQ_DONTWAIT) == 0)) {
if (!fManager.Interrupted() && ((flags & ZMQ_DONTWAIT) == 0)) {
if (timeout > 0) {
elapsed += fRcvTimeout;
if (elapsed >= timeout) {
@@ -365,20 +363,6 @@ void Socket::Close()
fSocket = nullptr;
}
void Socket::Interrupt()
{
Manager::Interrupt();
Message::fInterrupted = true;
fInterrupted = true;
}
void Socket::Resume()
{
Manager::Resume();
Message::fInterrupted = false;
fInterrupted = false;
}
void Socket::SetOption(const string& option, const void* value, size_t valueSize)
{
if (zmq_setsockopt(fSocket, GetConstant(option), value, valueSize) < 0) {

5
fairmq/shmem/Socket.h
View File

@@ -46,9 +46,6 @@ class Socket final : public fair::mq::Socket
void Close() override;
static void Interrupt();
static void Resume();
void SetOption(const std::string& option, const void* value, size_t valueSize) override;
void GetOption(const std::string& option, void* value, size_t* valueSize) override;
@@ -81,8 +78,6 @@ class Socket final : public fair::mq::Socket
std::atomic<unsigned long> fMessagesTx;
std::atomic<unsigned long> fMessagesRx;
static std::atomic<bool> fInterrupted;
int fSndTimeout;
int fRcvTimeout;
};

20
fairmq/shmem/TransportFactory.cxx
View File

@@ -164,6 +164,26 @@ UnmanagedRegionPtr TransportFactory::CreateUnmanagedRegion(const size_t size, Re
return tools::make_unique<UnmanagedRegion>(*fManager, size, callback, path, flags);
}
UnmanagedRegionPtr TransportFactory::CreateUnmanagedRegion(const size_t size, const int64_t userFlags, RegionCallback callback, const std::string& path /* = "" */, int flags /* = 0 */) const
{
return tools::make_unique<UnmanagedRegion>(*fManager, size, userFlags, callback, path, flags);
}
void TransportFactory::SubscribeToRegionEvents(RegionEventCallback callback)
{
fManager->SubscribeToRegionEvents(callback);
}
void TransportFactory::UnsubscribeFromRegionEvents()
{
fManager->UnsubscribeFromRegionEvents();
}
vector<fair::mq::RegionInfo> TransportFactory::GetRegionInfo()
{
return fManager->GetRegionInfo();
}
Transport TransportFactory::GetType() const
{
return fTransportType;

9
fairmq/shmem/TransportFactory.h
View File

@@ -50,11 +50,16 @@ class TransportFactory final : public fair::mq::TransportFactory
PollerPtr CreatePoller(const std::unordered_map<std::string, std::vector<FairMQChannel>>& channelsMap, const std::vector<std::string>& channelList) const override;
UnmanagedRegionPtr CreateUnmanagedRegion(const size_t size, RegionCallback callback = nullptr, const std::string& path = "", int flags = 0) const override;
UnmanagedRegionPtr CreateUnmanagedRegion(const size_t size, int64_t userFlags, RegionCallback callback = nullptr, const std::string& path = "", int flags = 0) const override;
void SubscribeToRegionEvents(RegionEventCallback callback) override;
void UnsubscribeFromRegionEvents() override;
std::vector<fair::mq::RegionInfo> GetRegionInfo() override;
Transport GetType() const override;
void Interrupt() override { Socket::Interrupt(); }
void Resume() override { Socket::Resume(); }
void Interrupt() override { fManager->Interrupt(); }
void Resume() override { fManager->Resume(); }
void Reset() override;
void IncrementMsgCounter() { ++fMsgCounter; }

52
fairmq/shmem/UnmanagedRegion.cxx
View File

@@ -1,52 +0,0 @@
/********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#include "Common.h"
#include "UnmanagedRegion.h"
using namespace std;
namespace bipc = ::boost::interprocess;
namespace fair
{
namespace mq
{
namespace shmem
{
UnmanagedRegion::UnmanagedRegion(Manager& manager, const size_t size, RegionCallback callback, const std::string& path /* = "" */, int flags /* = 0 */)
: fManager(manager)
, fRegion(nullptr)
, fRegionId(0)
{
try {
RegionCounter* rc = fManager.ManagementSegment().find<RegionCounter>(bipc::unique_instance).first;
if (rc) {
LOG(debug) << "region counter found, with value of " << rc->fCount << ". incrementing.";
(rc->fCount)++;
LOG(debug) << "incremented region counter, now: " << rc->fCount;
} else {
LOG(debug) << "no region counter found, creating one and initializing with 1";
rc = fManager.ManagementSegment().construct<RegionCounter>(bipc::unique_instance)(1);
LOG(debug) << "initialized region counter with: " << rc->fCount;
}
fRegionId = rc->fCount;
fRegion = fManager.CreateRegion(size, fRegionId, callback, path, flags);
} catch (bipc::interprocess_exception& e) {
LOG(error) << "cannot create region. Already created/not cleaned up?";
LOG(error) << e.what();
throw;
}
}
}
}
}

16
fairmq/shmem/UnmanagedRegion.h
View File

@@ -36,7 +36,19 @@ class UnmanagedRegion final : public fair::mq::UnmanagedRegion
friend class Socket;
public:
UnmanagedRegion(Manager& manager, const size_t size, RegionCallback callback = nullptr, const std::string& path = "", int flags = 0);
UnmanagedRegion(Manager& manager, const size_t size, RegionCallback callback, const std::string& path = "", int flags = 0)
: UnmanagedRegion(manager, size, 0, callback, path, flags)
{}
UnmanagedRegion(Manager& manager, const size_t size, const int64_t userFlags, RegionCallback callback, const std::string& path = "", int flags = 0)
: fManager(manager)
, fRegion(nullptr)
, fRegionId(0)
{
auto result = fManager.CreateRegion(size, userFlags, callback, path, flags);
fRegion = result.first;
fRegionId = result.second;
}
void* GetData() const override { return fRegion->get_address(); }
size_t GetSize() const override { return fRegion->get_size(); }
@@ -53,4 +65,4 @@ class UnmanagedRegion final : public fair::mq::UnmanagedRegion
}
}
#endif /* FAIR_MQ_SHMEM_UNMANAGEDREGION_H_ */
#endif /* FAIR_MQ_SHMEM_UNMANAGEDREGION_H_ */