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

FairMQ: Introduce callbacks for the FairMQUnmanagedRegion.

Browse Source 
Callbacks are called when the data buffer of the message assiciated
with the corresponding region is no longer needed by the transport.
Example in examples/advanced/Region/
This commit is contained in:
Alexey Rybalchenko 2017-11-14 17:00:37 +01:00 committed by Mohammad Al-Turany
parent 378c47c5e5
commit 58a312b730
39 changed files with 747 additions and 548 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
fairmq/CMakeLists.txt
View File

@ -102,8 +102,10 @@ set(FAIRMQ_HEADER_FILES
shmem/FairMQUnmanagedRegionSHM.h shmem/FairMQUnmanagedRegionSHM.h
shmem/FairMQSocketSHM.h shmem/FairMQSocketSHM.h
shmem/FairMQTransportFactorySHM.h shmem/FairMQTransportFactorySHM.h
shmem/FairMQShmMonitor.h shmem/Common.h
shmem/FairMQShmCommon.h shmem/Manager.h
shmem/Monitor.h
shmem/Region.h
tools/CppSTL.h tools/CppSTL.h
tools/Network.h tools/Network.h
tools/Strings.h tools/Strings.h
@ -163,7 +165,9 @@ set(FAIRMQ_SOURCE_FILES
shmem/FairMQUnmanagedRegionSHM.cxx shmem/FairMQUnmanagedRegionSHM.cxx
shmem/FairMQSocketSHM.cxx shmem/FairMQSocketSHM.cxx
shmem/FairMQTransportFactorySHM.cxx shmem/FairMQTransportFactorySHM.cxx
shmem/FairMQShmMonitor.cxx shmem/Manager.cxx
shmem/Monitor.cxx
shmem/Region.cxx
zeromq/FairMQMessageZMQ.cxx zeromq/FairMQMessageZMQ.cxx
zeromq/FairMQPollerZMQ.cxx zeromq/FairMQPollerZMQ.cxx
zeromq/FairMQUnmanagedRegionZMQ.cxx zeromq/FairMQUnmanagedRegionZMQ.cxx
@ -262,7 +266,7 @@ target_link_libraries(splitter FairMQ)
add_executable(runConfigExample options/runConfigEx.cxx) add_executable(runConfigExample options/runConfigEx.cxx)
target_link_libraries(runConfigExample FairMQ) target_link_libraries(runConfigExample FairMQ)
add_executable(shmmonitor shmem/runFairMQShmMonitor.cxx) add_executable(shmmonitor shmem/runMonitor.cxx)
target_link_libraries(shmmonitor FairMQ) target_link_libraries(shmmonitor FairMQ)

36
fairmq/FairMQDevice.cxx
View File

@ -119,7 +119,6 @@ void FairMQDevice::InitWrapper()
} }
FairMQMessagePtr msg(fTransportFactory->CreateMessage()); FairMQMessagePtr msg(fTransportFactory->CreateMessage());
msg->SetDeviceId(fId);
} }
// Containers to store the uninitialized channels. // Containers to store the uninitialized channels.
@ -131,18 +130,14 @@ void FairMQDevice::InitWrapper()
{ {
for (auto vi = mi.second.begin(); vi != mi.second.end(); ++vi) for (auto vi = mi.second.begin(); vi != mi.second.end(); ++vi)
{ {
if (vi->fModified) // if (vi->fModified)
{ // {
if (vi->fReset) // if (vi->fReset)
{ // {
vi->fSocket->Close(); // vi->fSocket.reset();
vi->fSocket = nullptr; // vi->fPoller.reset();
// vi->fChannelCmdSocket.reset();
vi->fPoller = nullptr; // }
vi->fChannelCmdSocket->Close();
vi->fChannelCmdSocket = nullptr;
}
// set channel name: name + vector index // set channel name: name + vector index
vi->fName = fair::mq::tools::ToString(mi.first, "[", vi - (mi.second).begin(), "]"); vi->fName = fair::mq::tools::ToString(mi.first, "[", vi - (mi.second).begin(), "]");
@ -176,7 +171,7 @@ void FairMQDevice::InitWrapper()
LOG(ERROR) << "Cannot update configuration. Socket method (bind/connect) not specified."; LOG(ERROR) << "Cannot update configuration. Socket method (bind/connect) not specified.";
throw runtime_error("Cannot update configuration. Socket method (bind/connect) not specified."); throw runtime_error("Cannot update configuration. Socket method (bind/connect) not specified.");
} }
} // }
} }
} }
@ -805,7 +800,6 @@ shared_ptr<FairMQTransportFactory> FairMQDevice::AddTransport(const string& tran
} }
FairMQMessagePtr msg(tr->CreateMessage()); FairMQMessagePtr msg(tr->CreateMessage());
msg->SetDeviceId(fId);
return move(tr); return move(tr);
} }
@ -1064,14 +1058,10 @@ void FairMQDevice::Reset()
// iterate over the channels vector // iterate over the channels vector
for (auto& vi : mi.second) for (auto& vi : mi.second)
{ {
vi.fReset = true; // vi.fReset = true;
// vi.fSocket->Close(); vi.fSocket.reset();
// vi.fSocket = nullptr; vi.fPoller.reset();
vi.fChannelCmdSocket.reset();
// vi.fPoller = nullptr;
// vi.fChannelCmdSocket->Close();
// vi.fChannelCmdSocket = nullptr;
} }
} }
} }

4
fairmq/FairMQDevice.h
View File

@ -240,9 +240,9 @@ class FairMQDevice : public FairMQStateMachine
return Transport()->CreateUnmanagedRegion(size); return Transport()->CreateUnmanagedRegion(size);
} }
FairMQUnmanagedRegionPtr NewUnmanagedRegionFor(const std::string& channel, int index, const size_t size) FairMQUnmanagedRegionPtr NewUnmanagedRegionFor(const std::string& channel, int index, const size_t size, FairMQRegionCallback callback = nullptr)
{ {
return fChannels.at(channel).at(index).Transport()->CreateUnmanagedRegion(size); return fChannels.at(channel).at(index).Transport()->CreateUnmanagedRegion(size, callback);
} }
template<typename ...Ts> template<typename ...Ts>

7
fairmq/FairMQMessage.h
View File

@ -1,8 +1,8 @@
/******************************************************************************** /********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH * * Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* * * *
* This software is distributed under the terms of the * * This software is distributed under the terms of the *
* GNU Lesser General Public Licence version 3 (LGPL) version 3, * * GNU Lesser General Public Licence version 3 (LGPL) version 3, *
* copied verbatim in the file "LICENSE" * * copied verbatim in the file "LICENSE" *
********************************************************************************/ ********************************************************************************/
/** /**
@ -32,9 +32,8 @@ class FairMQMessage
virtual void* GetMessage() = 0; virtual void* GetMessage() = 0;
virtual void* GetData() = 0; virtual void* GetData() = 0;
virtual size_t GetSize() = 0; virtual size_t GetSize() = 0;
virtual void SetMessage(void* data, size_t size) = 0;
virtual void SetDeviceId(const std::string& deviceId) = 0; virtual void SetMessage(void* data, size_t size) = 0;
virtual FairMQ::Transport GetType() const = 0; virtual FairMQ::Transport GetType() const = 0;

2
fairmq/FairMQTransportFactory.h
View File

@ -66,7 +66,7 @@ class FairMQTransportFactory
/// Create a poller for two sockets /// Create a poller for two sockets
virtual FairMQPollerPtr CreatePoller(const FairMQSocket& cmdSocket, const FairMQSocket& dataSocket) const = 0; virtual FairMQPollerPtr CreatePoller(const FairMQSocket& cmdSocket, const FairMQSocket& dataSocket) const = 0;
virtual FairMQUnmanagedRegionPtr CreateUnmanagedRegion(const size_t size) const = 0; virtual FairMQUnmanagedRegionPtr CreateUnmanagedRegion(const size_t size, FairMQRegionCallback callback = nullptr) const = 0;
/// Get transport type /// Get transport type
virtual FairMQ::Transport GetType() const = 0; virtual FairMQ::Transport GetType() const = 0;

5
fairmq/FairMQUnmanagedRegion.h
View File

@ -10,7 +10,10 @@
#define FAIRMQUNMANAGEDREGION_H_ #define FAIRMQUNMANAGEDREGION_H_
#include <cstddef> // size_t #include <cstddef> // size_t
#include <memory> // unique_ptr #include <memory> // std::unique_ptr
#include <functional> // std::function
using FairMQRegionCallback = std::function<void(void*, size_t)>;
class FairMQUnmanagedRegion class FairMQUnmanagedRegion
{ {

16
fairmq/nanomsg/FairMQMessageNN.cxx
View File

@ -22,14 +22,13 @@
using namespace std; using namespace std;
string FairMQMessageNN::fDeviceID = string();
FairMQ::Transport FairMQMessageNN::fTransportType = FairMQ::Transport::NN; FairMQ::Transport FairMQMessageNN::fTransportType = FairMQ::Transport::NN;
FairMQMessageNN::FairMQMessageNN() FairMQMessageNN::FairMQMessageNN()
: fMessage(nullptr) : fMessage(nullptr)
, fSize(0) , fSize(0)
, fReceiving(false) , fReceiving(false)
, fRegion(false) , fRegionPtr(nullptr)
{ {
fMessage = nn_allocmsg(0, 0); fMessage = nn_allocmsg(0, 0);
if (!fMessage) if (!fMessage)
@ -42,7 +41,7 @@ FairMQMessageNN::FairMQMessageNN(const size_t size)
: fMessage(nullptr) : fMessage(nullptr)
, fSize(0) , fSize(0)
, fReceiving(false) , fReceiving(false)
, fRegion(false) , fRegionPtr(nullptr)
{ {
fMessage = nn_allocmsg(size, 0); fMessage = nn_allocmsg(size, 0);
if (!fMessage) if (!fMessage)
@ -62,7 +61,7 @@ FairMQMessageNN::FairMQMessageNN(void* data, const size_t size, fairmq_free_fn*
: fMessage(nullptr) : fMessage(nullptr)
, fSize(0) , fSize(0)
, fReceiving(false) , fReceiving(false)
, fRegion(false) , fRegionPtr(nullptr)
{ {
fMessage = nn_allocmsg(size, 0); fMessage = nn_allocmsg(size, 0);
if (!fMessage) if (!fMessage)
@ -84,11 +83,11 @@ FairMQMessageNN::FairMQMessageNN(void* data, const size_t size, fairmq_free_fn*
} }
} }
FairMQMessageNN::FairMQMessageNN(FairMQUnmanagedRegionPtr& /*region*/, void* data, const size_t size) FairMQMessageNN::FairMQMessageNN(FairMQUnmanagedRegionPtr& region, void* data, const size_t size)
: fMessage(data) : fMessage(data)
, fSize(size) , fSize(size)
, fReceiving(false) , fReceiving(false)
, fRegion(true) , fRegionPtr(region.get())
{ {
// currently nanomsg will copy the buffer (data) inside nn_sendmsg() // currently nanomsg will copy the buffer (data) inside nn_sendmsg()
} }
@ -153,11 +152,6 @@ void FairMQMessageNN::SetMessage(void* data, const size_t size)
fSize = size; fSize = size;
} }
void FairMQMessageNN::SetDeviceId(const string& deviceId)
{
fDeviceID = deviceId;
}
FairMQ::Transport FairMQMessageNN::GetType() const FairMQ::Transport FairMQMessageNN::GetType() const
{ {
return fTransportType; return fTransportType;

6
fairmq/nanomsg/FairMQMessageNN.h
View File

@ -17,6 +17,7 @@
#include <cstddef> #include <cstddef>
#include <string> #include <string>
#include <memory>
#include "FairMQMessage.h" #include "FairMQMessage.h"
#include "FairMQUnmanagedRegion.h" #include "FairMQUnmanagedRegion.h"
@ -42,8 +43,6 @@ class FairMQMessageNN : public FairMQMessage
virtual void SetMessage(void* data, const size_t size); virtual void SetMessage(void* data, const size_t size);
virtual void SetDeviceId(const std::string& deviceId);
virtual FairMQ::Transport GetType() const; virtual FairMQ::Transport GetType() const;
virtual void Copy(const std::unique_ptr<FairMQMessage>& msg); virtual void Copy(const std::unique_ptr<FairMQMessage>& msg);
@ -56,8 +55,7 @@ class FairMQMessageNN : public FairMQMessage
void* fMessage; void* fMessage;
size_t fSize; size_t fSize;
bool fReceiving; bool fReceiving;
bool fRegion; FairMQUnmanagedRegion* fRegionPtr;
static std::string fDeviceID;
static FairMQ::Transport fTransportType; static FairMQ::Transport fTransportType;
void Clear(); void Clear();

8
fairmq/nanomsg/FairMQSocketNN.cxx
View File

@ -15,6 +15,7 @@
#include "FairMQSocketNN.h" #include "FairMQSocketNN.h"
#include "FairMQMessageNN.h" #include "FairMQMessageNN.h"
#include "FairMQLogger.h" #include "FairMQLogger.h"
#include "FairMQUnmanagedRegionNN.h"
#include <nanomsg/nn.h> #include <nanomsg/nn.h>
#include <nanomsg/pipeline.h> #include <nanomsg/pipeline.h>
@ -127,14 +128,17 @@ int FairMQSocketNN::Send(FairMQMessagePtr& msg, const int flags)
while (true) while (true)
{ {
void* ptr = msg->GetMessage(); void* ptr = msg->GetMessage();
if (static_cast<FairMQMessageNN*>(msg.get())->fRegion == false) if (static_cast<FairMQMessageNN*>(msg.get())->fRegionPtr == nullptr)
{ {
nbytes = nn_send(fSocket, &ptr, NN_MSG, flags); nbytes = nn_send(fSocket, &ptr, NN_MSG, flags);
} }
else else
{ {
nbytes = nn_send(fSocket, ptr, msg->GetSize(), flags); nbytes = nn_send(fSocket, ptr, msg->GetSize(), flags);
// nn_send copies the data, safe to call region callback here
static_cast<FairMQUnmanagedRegionNN*>(static_cast<FairMQMessageNN*>(msg.get())->fRegionPtr)->fCallback(msg->GetMessage(), msg->GetSize());
} }
if (nbytes >= 0) if (nbytes >= 0)
{ {
fBytesTx += nbytes; fBytesTx += nbytes;
@ -239,6 +243,8 @@ int64_t FairMQSocketNN::Send(vector<unique_ptr<FairMQMessage>>& msgVec, const in
static_cast<FairMQMessageNN*>(msgVec[i].get())->fReceiving = false; static_cast<FairMQMessageNN*>(msgVec[i].get())->fReceiving = false;
packer.pack_bin(msgVec[i]->GetSize()); packer.pack_bin(msgVec[i]->GetSize());
packer.pack_bin_body(static_cast<char*>(msgVec[i]->GetData()), msgVec[i]->GetSize()); packer.pack_bin_body(static_cast<char*>(msgVec[i]->GetData()), msgVec[i]->GetSize());
// call region callback
static_cast<FairMQUnmanagedRegionNN*>(static_cast<FairMQMessageNN*>(msgVec[i].get())->fRegionPtr)->fCallback(msgVec[i]->GetMessage(), msgVec[i]->GetSize());
} }
int64_t nbytes = -1; int64_t nbytes = -1;

4
fairmq/nanomsg/FairMQTransportFactoryNN.cxx
View File

@ -65,9 +65,9 @@ FairMQPollerPtr FairMQTransportFactoryNN::CreatePoller(const FairMQSocket& cmdSo
return unique_ptr<FairMQPoller>(new FairMQPollerNN(cmdSocket, dataSocket)); return unique_ptr<FairMQPoller>(new FairMQPollerNN(cmdSocket, dataSocket));
} }
FairMQUnmanagedRegionPtr FairMQTransportFactoryNN::CreateUnmanagedRegion(const size_t size) const FairMQUnmanagedRegionPtr FairMQTransportFactoryNN::CreateUnmanagedRegion(const size_t size, FairMQRegionCallback callback) const
{ {
return unique_ptr<FairMQUnmanagedRegion>(new FairMQUnmanagedRegionNN(size)); return unique_ptr<FairMQUnmanagedRegion>(new FairMQUnmanagedRegionNN(size, callback));
} }
FairMQ::Transport FairMQTransportFactoryNN::GetType() const FairMQ::Transport FairMQTransportFactoryNN::GetType() const

2
fairmq/nanomsg/FairMQTransportFactoryNN.h
View File

@ -37,7 +37,7 @@ class FairMQTransportFactoryNN : public FairMQTransportFactory
FairMQPollerPtr CreatePoller(const std::unordered_map<std::string, std::vector<FairMQChannel>>& channelsMap, const std::vector<std::string>& channelList) const override; FairMQPollerPtr CreatePoller(const std::unordered_map<std::string, std::vector<FairMQChannel>>& channelsMap, const std::vector<std::string>& channelList) const override;
FairMQPollerPtr CreatePoller(const FairMQSocket& cmdSocket, const FairMQSocket& dataSocket) const override; FairMQPollerPtr CreatePoller(const FairMQSocket& cmdSocket, const FairMQSocket& dataSocket) const override;
FairMQUnmanagedRegionPtr CreateUnmanagedRegion(const size_t size) const override; FairMQUnmanagedRegionPtr CreateUnmanagedRegion(const size_t size, FairMQRegionCallback callback) const override;
FairMQ::Transport GetType() const override; FairMQ::Transport GetType() const override;

3
fairmq/nanomsg/FairMQUnmanagedRegionNN.cxx
View File

@ -11,9 +11,10 @@
using namespace std; using namespace std;
FairMQUnmanagedRegionNN::FairMQUnmanagedRegionNN(const size_t size) FairMQUnmanagedRegionNN::FairMQUnmanagedRegionNN(const size_t size, FairMQRegionCallback callback)
: fBuffer(malloc(size)) : fBuffer(malloc(size))
, fSize(size) , fSize(size)
, fCallback(callback)
{ {
} }

3
fairmq/nanomsg/FairMQUnmanagedRegionNN.h
View File

@ -18,7 +18,7 @@ class FairMQUnmanagedRegionNN : public FairMQUnmanagedRegion
friend class FairMQSocketNN; friend class FairMQSocketNN;
public: public:
FairMQUnmanagedRegionNN(const size_t size); FairMQUnmanagedRegionNN(const size_t size, FairMQRegionCallback callback);
FairMQUnmanagedRegionNN(const FairMQUnmanagedRegionNN&) = delete; FairMQUnmanagedRegionNN(const FairMQUnmanagedRegionNN&) = delete;
FairMQUnmanagedRegionNN operator=(const FairMQUnmanagedRegionNN&) = delete; FairMQUnmanagedRegionNN operator=(const FairMQUnmanagedRegionNN&) = delete;
@ -30,6 +30,7 @@ class FairMQUnmanagedRegionNN : public FairMQUnmanagedRegion
private: private:
void* fBuffer; void* fBuffer;
size_t fSize; size_t fSize;
FairMQRegionCallback fCallback;
}; };
#endif /* FAIRMQUNMANAGEDREGIONNN_H_ */ #endif /* FAIRMQUNMANAGEDREGIONNN_H_ */

30
fairmq/options/FairMQProgOptions.cxx
View File

@ -322,21 +322,21 @@ int FairMQProgOptions::NotifySwitchOption()
void FairMQProgOptions::FillOptionDescription(boost::program_options::options_description& options) void FairMQProgOptions::FillOptionDescription(boost::program_options::options_description& options)
{ {
options.add_options() options.add_options()
("id", po::value<string>(), "Device ID (required argument).") ("id", po::value<string>(), "Device ID (required argument).")
("io-threads", po::value<int >()->default_value(1), "Number of I/O threads.") ("io-threads", po::value<int >()->default_value(1), "Number of I/O threads.")
("transport", po::value<string>()->default_value("zeromq"), "Transport ('zeromq'/'nanomsg').") ("transport", po::value<string>()->default_value("zeromq"), "Transport ('zeromq'/'nanomsg').")
("config", po::value<string>()->default_value("static"), "Config source ('static'/<config library filename>).") ("config", po::value<string>()->default_value("static"), "Config source ('static'/<config library filename>).")
("network-interface", po::value<string>()->default_value("default"), "Network interface to bind on (e.g. eth0, ib0..., default will try to detect the interface of the default route).") ("network-interface", po::value<string>()->default_value("default"), "Network interface to bind on (e.g. eth0, ib0..., default will try to detect the interface of the default route).")
("config-key", po::value<string>(), "Use provided value instead of device id for fetching the configuration from the config file.") ("config-key", po::value<string>(), "Use provided value instead of device id for fetching the configuration from the config file.")
("catch-signals", po::value<int >()->default_value(1), "Enable signal handling (1/0).") ("catch-signals", po::value<int >()->default_value(1), "Enable signal handling (1/0).")
("initialization-timeout", po::value<int >()->default_value(120), "Timeout for the initialization in seconds (when expecting dynamic initialization).") ("initialization-timeout", po::value<int >()->default_value(120), "Timeout for the initialization in seconds (when expecting dynamic initialization).")
("port-range-min", po::value<int >()->default_value(22000), "Start of the port range for dynamic initialization.") ("port-range-min", po::value<int >()->default_value(22000), "Start of the port range for dynamic initialization.")
("port-range-max", po::value<int >()->default_value(32000), "End of the port range for dynamic initialization.") ("port-range-max", po::value<int >()->default_value(32000), "End of the port range for dynamic initialization.")
("log-to-file", po::value<string>()->default_value(""), "Log output to a file.") ("log-to-file", po::value<string>()->default_value(""), "Log output to a file.")
("print-channels", po::value<bool >()->implicit_value(true), "Print registered channel endpoints in a machine-readable format (<channel name>:<min num subchannels>:<max num subchannels>)") ("print-channels", po::value<bool >()->implicit_value(true), "Print registered channel endpoints in a machine-readable format (<channel name>:<min num subchannels>:<max num subchannels>)")
("shm-segment-size", po::value<size_t>()->default_value(2000000000), "shmem transport: size of the shared memory segment (in bytes).") ("shm-segment-size", po::value<size_t>()->default_value(2000000000), "shmem transport: size of the shared memory segment (in bytes).")
("shm-segment-name", po::value<string>()->default_value("fairmq_shmem_main"), "shmem transport: name of the shared memory segment.") ("shm-segment-name", po::value<string>()->default_value("fmq_shm_main"), "shmem transport: name of the shared memory segment.")
("rate", po::value<float>()->default_value(0.), "rate for conditional run loop (Hz)") ("rate", po::value<float >()->default_value(0.), "rate for conditional run loop (Hz)")
; ;
} }

20
fairmq/shmem/FairMQShmCommon.h → fairmq/shmem/Common.h
View File

@ -1,8 +1,8 @@
/******************************************************************************** /********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH * * Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* * * *
* This software is distributed under the terms of the * * This software is distributed under the terms of the *
* GNU Lesser General Public Licence version 3 (LGPL) version 3, * * GNU Lesser General Public Licence version 3 (LGPL) version 3, *
* copied verbatim in the file "LICENSE" * * copied verbatim in the file "LICENSE" *
********************************************************************************/ ********************************************************************************/
#ifndef FAIR_MQ_SHMEM_COMMON_H_ #ifndef FAIR_MQ_SHMEM_COMMON_H_
@ -53,6 +53,22 @@ struct alignas(32) MetaHeader
boost::interprocess::managed_shared_memory::handle_t fHandle; boost::interprocess::managed_shared_memory::handle_t fHandle;
}; };
struct RegionBlock
{
RegionBlock()
: fHandle()
, fSize(0)
{}
RegionBlock(boost::interprocess::managed_shared_memory::handle_t handle, size_t size)
: fHandle(handle)
, fSize(size)
{}
boost::interprocess::managed_shared_memory::handle_t fHandle;
size_t fSize;
};
} // namespace shmem } // namespace shmem
} // namespace mq } // namespace mq
} // namespace fair } // namespace fair

115
fairmq/shmem/FairMQMessageSHM.cxx
View File

@ -8,28 +8,31 @@
#include <string> #include <string>
#include <cstdlib> #include <cstdlib>
#include <boost/date_time/posix_time/posix_time.hpp>
#include "FairMQMessageSHM.h" #include "FairMQMessageSHM.h"
#include "FairMQUnmanagedRegionSHM.h" #include "FairMQUnmanagedRegionSHM.h"
#include "FairMQLogger.h" #include "FairMQLogger.h"
#include "FairMQShmCommon.h" #include "Common.h"
using namespace std; using namespace std;
using namespace fair::mq::shmem; using namespace fair::mq::shmem;
namespace bipc = boost::interprocess; namespace bipc = boost::interprocess;
namespace bpt = boost::posix_time;
atomic<bool> FairMQMessageSHM::fInterrupted(false); atomic<bool> FairMQMessageSHM::fInterrupted(false);
FairMQ::Transport FairMQMessageSHM::fTransportType = FairMQ::Transport::SHM; FairMQ::Transport FairMQMessageSHM::fTransportType = FairMQ::Transport::SHM;
FairMQMessageSHM::FairMQMessageSHM() FairMQMessageSHM::FairMQMessageSHM(Manager& manager)
: fMessage() : fManager(manager)
, fMessage()
, fQueued(false) , fQueued(false)
, fMetaCreated(false) , fMetaCreated(false)
, fRegionId(0) , fRegionId(0)
, fHandle() , fHandle(-1)
, fSize(0) , fSize(0)
, fLocalPtr(nullptr) , fLocalPtr(nullptr)
, fRemoteRegion(nullptr)
{ {
if (zmq_msg_init(&fMessage) != 0) if (zmq_msg_init(&fMessage) != 0)
{ {
@ -38,28 +41,28 @@ FairMQMessageSHM::FairMQMessageSHM()
fMetaCreated = true; fMetaCreated = true;
} }
FairMQMessageSHM::FairMQMessageSHM(const size_t size) FairMQMessageSHM::FairMQMessageSHM(Manager& manager, const size_t size)
: fMessage() : fManager(manager)
, fMessage()
, fQueued(false) , fQueued(false)
, fMetaCreated(false) , fMetaCreated(false)
, fRegionId(0) , fRegionId(0)
, fHandle() , fHandle(-1)
, fSize(0) , fSize(0)
, fLocalPtr(nullptr) , fLocalPtr(nullptr)
, fRemoteRegion(nullptr)
{ {
InitializeChunk(size); InitializeChunk(size);
} }
FairMQMessageSHM::FairMQMessageSHM(void* data, const size_t size, fairmq_free_fn* ffn, void* hint) FairMQMessageSHM::FairMQMessageSHM(Manager& manager, void* data, const size_t size, fairmq_free_fn* ffn, void* hint)
: fMessage() : fManager(manager)
, fMessage()
, fQueued(false) , fQueued(false)
, fMetaCreated(false) , fMetaCreated(false)
, fRegionId(0) , fRegionId(0)
, fHandle() , fHandle(-1)
, fSize(0) , fSize(0)
, fLocalPtr(nullptr) , fLocalPtr(nullptr)
, fRemoteRegion(nullptr)
{ {
if (InitializeChunk(size)) if (InitializeChunk(size))
{ {
@ -75,15 +78,15 @@ FairMQMessageSHM::FairMQMessageSHM(void* data, const size_t size, fairmq_free_fn
} }
} }
FairMQMessageSHM::FairMQMessageSHM(FairMQUnmanagedRegionPtr& region, void* data, const size_t size) FairMQMessageSHM::FairMQMessageSHM(Manager& manager, FairMQUnmanagedRegionPtr& region, void* data, const size_t size)
: fMessage() : fManager(manager)
, fMessage()
, fQueued(false) , fQueued(false)
, fMetaCreated(false) , fMetaCreated(false)
, fRegionId(static_cast<FairMQUnmanagedRegionSHM*>(region.get())->fRegionId) , fRegionId(static_cast<FairMQUnmanagedRegionSHM*>(region.get())->fRegionId)
, fHandle() , fHandle(-1)
, fSize(size) , fSize(size)
, fLocalPtr(data) , fLocalPtr(data)
, fRemoteRegion(nullptr)
{ {
fHandle = (bipc::managed_shared_memory::handle_t)(reinterpret_cast<const char*>(data) - reinterpret_cast<const char*>(region->GetData())); fHandle = (bipc::managed_shared_memory::handle_t)(reinterpret_cast<const char*>(data) - reinterpret_cast<const char*>(region->GetData()));
@ -105,11 +108,11 @@ FairMQMessageSHM::FairMQMessageSHM(FairMQUnmanagedRegionPtr& region, void* data,
bool FairMQMessageSHM::InitializeChunk(const size_t size) bool FairMQMessageSHM::InitializeChunk(const size_t size)
{ {
while (!fHandle) while (fHandle < 0)
{ {
try try
{ {
fLocalPtr = Manager::Instance().Segment()->allocate(size); fLocalPtr = fManager.Segment().allocate(size);
} }
catch (bipc::bad_alloc& ba) catch (bipc::bad_alloc& ba)
{ {
@ -124,7 +127,7 @@ bool FairMQMessageSHM::InitializeChunk(const size_t size)
continue; continue;
} }
} }
fHandle = Manager::Instance().Segment()->get_handle_from_address(fLocalPtr); fHandle = fManager.Segment().get_handle_from_address(fLocalPtr);
} }
fSize = size; fSize = size;
@ -202,15 +205,20 @@ void* FairMQMessageSHM::GetData()
{ {
if (fRegionId == 0) if (fRegionId == 0)
{ {
return Manager::Instance().Segment()->get_address_from_handle(fHandle); return fManager.Segment().get_address_from_handle(fHandle);
} }
else else
{ {
if (!fRemoteRegion) boost::interprocess::mapped_region* region = fManager.GetRemoteRegion(fRegionId);
if (region)
{ {
fRemoteRegion = FairMQUnmanagedRegionSHM::GetRemoteRegion(fRegionId); fLocalPtr = reinterpret_cast<char*>(region->get_address()) + fHandle;
}
else
{
// LOG(WARN) << "could not get pointer from a region message";
fLocalPtr = nullptr;
} }
fLocalPtr = reinterpret_cast<char*>(fRemoteRegion->get_address()) + fHandle;
return fLocalPtr; return fLocalPtr;
} }
} }
@ -226,11 +234,6 @@ void FairMQMessageSHM::SetMessage(void*, const size_t)
// dummy method to comply with the interface. functionality not allowed in zeromq. // dummy method to comply with the interface. functionality not allowed in zeromq.
} }
void FairMQMessageSHM::SetDeviceId(const string& /*deviceId*/)
{
// fDeviceID = deviceId;
}
FairMQ::Transport FairMQMessageSHM::GetType() const FairMQ::Transport FairMQMessageSHM::GetType() const
{ {
return fTransportType; return fTransportType;
@ -238,7 +241,7 @@ FairMQ::Transport FairMQMessageSHM::GetType() const
void FairMQMessageSHM::Copy(const unique_ptr<FairMQMessage>& msg) void FairMQMessageSHM::Copy(const unique_ptr<FairMQMessage>& msg)
{ {
if (!fHandle) if (fHandle < 0)
{ {
bipc::managed_shared_memory::handle_t otherHandle = static_cast<FairMQMessageSHM*>(msg.get())->fHandle; bipc::managed_shared_memory::handle_t otherHandle = static_cast<FairMQMessageSHM*>(msg.get())->fHandle;
if (otherHandle) if (otherHandle)
@ -261,10 +264,56 @@ void FairMQMessageSHM::Copy(const unique_ptr<FairMQMessage>& msg)
void FairMQMessageSHM::CloseMessage() void FairMQMessageSHM::CloseMessage()
{ {
if (fHandle && !fQueued && fRegionId == 0) if (fHandle >= 0 && !fQueued)
{ {
Manager::Instance().Segment()->deallocate(Manager::Instance().Segment()->get_address_from_handle(fHandle)); if (fRegionId == 0)
fHandle = 0; {
fManager.Segment().deallocate(fManager.Segment().get_address_from_handle(fHandle));
fHandle = 0;
}
else
{
// send notification back to the receiver
// RegionBlock block(fHandle, fSize);
// if (fManager.GetRegionQueue(fRegionId).try_send(static_cast<void*>(&block), sizeof(RegionBlock), 0))
// {
// // LOG(INFO) << "true";
// }
// // else
// // {
// // LOG(DEBUG) << "could not send ack";
// // }
// timed version
RegionBlock block(fHandle, fSize);
bool success = false;
do
{
auto sndTill = bpt::microsec_clock::universal_time() + bpt::milliseconds(200);
bipc::message_queue* q = fManager.GetRegionQueue(fRegionId);
if (q)
{
if (q->timed_send(&block, sizeof(RegionBlock), 0, sndTill))
{
success = true;
}
else
{
if (fInterrupted)
{
break;
}
LOG(DEBUG) << "region ack queue is full, retrying...";
}
}
else
{
// LOG(WARN) << "region ack queue for id " << fRegionId << " no longer exist. Not sending ack";
success = true;
}
}
while (!success);
}
} }
if (fMetaCreated) if (fMetaCreated)

36
fairmq/shmem/FairMQMessageSHM.h
View File

@ -18,54 +18,52 @@
#include "FairMQMessage.h" #include "FairMQMessage.h"
#include "FairMQUnmanagedRegion.h" #include "FairMQUnmanagedRegion.h"
#include "FairMQShmManager.h" #include "Manager.h"
class FairMQMessageSHM : public FairMQMessage class FairMQMessageSHM : public FairMQMessage
{ {
friend class FairMQSocketSHM; friend class FairMQSocketSHM;
public: public:
FairMQMessageSHM(); FairMQMessageSHM(fair::mq::shmem::Manager& manager);
FairMQMessageSHM(const size_t size); FairMQMessageSHM(fair::mq::shmem::Manager& manager, const size_t size);
FairMQMessageSHM(void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr); FairMQMessageSHM(fair::mq::shmem::Manager& manager, void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr);
FairMQMessageSHM(FairMQUnmanagedRegionPtr& region, void* data, const size_t size); FairMQMessageSHM(fair::mq::shmem::Manager& manager, FairMQUnmanagedRegionPtr& region, void* data, const size_t size);
FairMQMessageSHM(const FairMQMessageSHM&) = delete; FairMQMessageSHM(const FairMQMessageSHM&) = delete;
FairMQMessageSHM operator=(const FairMQMessageSHM&) = delete; FairMQMessageSHM operator=(const FairMQMessageSHM&) = delete;
bool InitializeChunk(const size_t size); bool InitializeChunk(const size_t size);
virtual void Rebuild(); void Rebuild() override;
virtual void Rebuild(const size_t size); void Rebuild(const size_t size) override;
virtual void Rebuild(void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr); void Rebuild(void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr) override;
virtual void* GetMessage(); void* GetMessage() override;
virtual void* GetData(); void* GetData() override;
virtual size_t GetSize(); size_t GetSize() override;
virtual void SetMessage(void* data, const size_t size); void SetMessage(void* data, const size_t size) override;
virtual void SetDeviceId(const std::string& deviceId); FairMQ::Transport GetType() const override;
virtual FairMQ::Transport GetType() const; void Copy(const std::unique_ptr<FairMQMessage>& msg) override;
virtual void Copy(const std::unique_ptr<FairMQMessage>& msg);
void CloseMessage(); void CloseMessage();
virtual ~FairMQMessageSHM(); ~FairMQMessageSHM() override;
private: private:
fair::mq::shmem::Manager& fManager;
zmq_msg_t fMessage; zmq_msg_t fMessage;
bool fQueued; bool fQueued;
bool fMetaCreated; bool fMetaCreated;
static std::atomic<bool> fInterrupted; static std::atomic<bool> fInterrupted;
static FairMQ::Transport fTransportType; static FairMQ::Transport fTransportType;
uint64_t fRegionId; uint64_t fRegionId;
bipc::managed_shared_memory::handle_t fHandle; boost::interprocess::managed_shared_memory::handle_t fHandle;
size_t fSize; size_t fSize;
void* fLocalPtr; void* fLocalPtr;
boost::interprocess::mapped_region* fRemoteRegion; // cache region ptr
}; };
#endif /* FAIRMQMESSAGESHM_H_ */ #endif /* FAIRMQMESSAGESHM_H_ */

16
fairmq/shmem/FairMQPollerSHM.h
View File

@ -1,8 +1,8 @@
/******************************************************************************** /********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH * * Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* * * *
* This software is distributed under the terms of the * * This software is distributed under the terms of the *
* GNU Lesser General Public Licence version 3 (LGPL) version 3, * * GNU Lesser General Public Licence version 3 (LGPL) version 3, *
* copied verbatim in the file "LICENSE" * * copied verbatim in the file "LICENSE" *
********************************************************************************/ ********************************************************************************/
#ifndef FAIRMQPOLLERSHM_H_ #ifndef FAIRMQPOLLERSHM_H_
@ -35,13 +35,13 @@ class FairMQPollerSHM : public FairMQPoller
void SetItemEvents(zmq_pollitem_t& item, const int type); void SetItemEvents(zmq_pollitem_t& item, const int type);
virtual void Poll(const int timeout); void Poll(const int timeout) override;
virtual bool CheckInput(const int index); bool CheckInput(const int index) override;
virtual bool CheckOutput(const int index); bool CheckOutput(const int index) override;
virtual bool CheckInput(const std::string channelKey, const int index); bool CheckInput(const std::string channelKey, const int index) override;
virtual bool CheckOutput(const std::string channelKey, const int index); bool CheckOutput(const std::string channelKey, const int index) override;
virtual ~FairMQPollerSHM(); ~FairMQPollerSHM() override;
private: private:
FairMQPollerSHM(const FairMQSocket& cmdSocket, const FairMQSocket& dataSocket); FairMQPollerSHM(const FairMQSocket& cmdSocket, const FairMQSocket& dataSocket);

217
fairmq/shmem/FairMQShmManager.h
View File

@ -1,217 +0,0 @@
/********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence version 3 (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
/**
* FairMQShmManager.h
*
* @since 2016-04-08
* @author A. Rybalchenko
*/
#ifndef FAIRMQSHMMANAGER_H_
#define FAIRMQSHMMANAGER_H_
#include <thread>
#include <chrono>
#include <boost/interprocess/managed_shared_memory.hpp>
#include <boost/interprocess/smart_ptr/shared_ptr.hpp>
#include "FairMQLogger.h"
namespace bipc = boost::interprocess;
namespace fair
{
namespace mq
{
namespace shmem
{
class Manager
{
public:
static Manager& Instance()
{
static Manager man;
return man;
}
void InitializeSegment(const std::string& op, const std::string& name, const size_t size = 0)
{
if (!fSegment)
{
try
{
if (op == "open_or_create")
{
fSegment = new bipc::managed_shared_memory(bipc::open_or_create, name.c_str(), size);
}
else if (op == "create_only")
{
fSegment = new bipc::managed_shared_memory(bipc::create_only, name.c_str(), size);
}
else if (op == "open_only")
{
int numTries = 0;
bool success = false;
do
{
try
{
fSegment = new bipc::managed_shared_memory(bipc::open_only, name.c_str());
success = true;
}
catch (bipc::interprocess_exception& ie)
{
if (++numTries == 5)
{
LOG(ERROR) << "Could not open shared memory after " << numTries << " attempts, exiting!";
exit(EXIT_FAILURE);
}
else
{
LOG(DEBUG) << "Could not open shared memory segment on try " << numTries << ". Retrying in 1 second...";
LOG(DEBUG) << ie.what();
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
}
}
}
while (!success);
}
else
{
LOG(ERROR) << "Unknown operation when initializing shared memory segment: " << op;
}
}
catch (std::exception& e)
{
LOG(ERROR) << "Exception during shared memory segment initialization: " << e.what() << ", application will now exit";
exit(EXIT_FAILURE);
}
}
else
{
LOG(INFO) << "Segment already initialized";
}
}
bipc::managed_shared_memory* Segment() const
{
if (fSegment)
{
return fSegment;
}
else
{
LOG(ERROR) << "Segment not initialized";
exit(EXIT_FAILURE);
}
}
void Remove()
{
if (bipc::shared_memory_object::remove("fairmq_shmem_main"))
{
LOG(DEBUG) << "shmem: successfully removed \"fairmq_shmem_main\" segment after the device has stopped.";
}
else
{
LOG(DEBUG) << "shmem: did not remove \"fairmq_shmem_main\" segment after the device stopped. Already removed?";
}
if (bipc::shared_memory_object::remove("fairmq_shmem_management"))
{
LOG(DEBUG) << "shmem: successfully removed \"fairmq_shmem_management\" segment after the device has stopped.";
}
else
{
LOG(DEBUG) << "shmem: did not remove \"fairmq_shmem_management\" segment after the device stopped. Already removed?";
}
}
bipc::managed_shared_memory& ManagementSegment()
{
return fManagementSegment;
}
private:
Manager()
: fSegment(nullptr)
, fManagementSegment(bipc::open_or_create, "fairmq_shmem_management", 65536)
{}
Manager(const Manager&) = delete;
Manager operator=(const Manager&) = delete;
bipc::managed_shared_memory* fSegment;
bipc::managed_shared_memory fManagementSegment;
};
// class Chunk
// {
// public:
// Chunk()
// : fHandle()
// , fSize(0)
// {
// }
// Chunk(const size_t size)
// : fHandle()
// , fSize(size)
// {
// void* ptr = Manager::Instance().Segment()->allocate(size);
// fHandle = Manager::Instance().Segment()->get_handle_from_address(ptr);
// }
// ~Chunk()
// {
// Manager::Instance().Segment()->deallocate(Manager::Instance().Segment()->get_address_from_handle(fHandle));
// }
// bipc::managed_shared_memory::handle_t GetHandle() const
// {
// return fHandle;
// }
// void* GetData() const
// {
// return Manager::Instance().Segment()->get_address_from_handle(fHandle);
// }
// size_t GetSize() const
// {
// return fSize;
// }
// private:
// bipc::managed_shared_memory::handle_t fHandle;
// size_t fSize;
// };
// typedef bipc::managed_shared_ptr<Chunk, bipc::managed_shared_memory>::type ShPtrType;
// struct ShPtrOwner
// {
// ShPtrOwner(const ShPtrType& other)
// : fPtr(other)
// {}
// ShPtrOwner(const ShPtrOwner& other)
// : fPtr(other.fPtr)
// {}
// ShPtrType fPtr;
// };
} // namespace shmem
} // namespace mq
} // namespace fair
#endif /* FAIRMQSHMMANAGER_H_ */

21
fairmq/shmem/FairMQSocketSHM.cxx
View File

@ -13,16 +13,17 @@
#include "FairMQMessageSHM.h" #include "FairMQMessageSHM.h"
#include "FairMQUnmanagedRegionSHM.h" #include "FairMQUnmanagedRegionSHM.h"
#include "FairMQLogger.h" #include "FairMQLogger.h"
#include "FairMQShmCommon.h" #include "Common.h"
using namespace std; using namespace std;
using namespace fair::mq::shmem; using namespace fair::mq::shmem;
atomic<bool> FairMQSocketSHM::fInterrupted(false); atomic<bool> FairMQSocketSHM::fInterrupted(false);
FairMQSocketSHM::FairMQSocketSHM(const string& type, const string& name, const string& id /*= ""*/, void* context) FairMQSocketSHM::FairMQSocketSHM(Manager& manager, const string& type, const string& name, const string& id /*= ""*/, void* context)
: FairMQSocket(ZMQ_SNDMORE, ZMQ_RCVMORE, ZMQ_DONTWAIT) : FairMQSocket(ZMQ_SNDMORE, ZMQ_RCVMORE, ZMQ_DONTWAIT)
, fSocket(NULL) , fManager(manager)
, fSocket(nullptr)
, fId() , fId()
, fBytesTx(0) , fBytesTx(0)
, fBytesRx(0) , fBytesRx(0)
@ -34,7 +35,7 @@ FairMQSocketSHM::FairMQSocketSHM(const string& type, const string& name, const s
assert(context); assert(context);
fSocket = zmq_socket(context, GetConstant(type)); fSocket = zmq_socket(context, GetConstant(type));
if (fSocket == NULL) if (fSocket == nullptr)
{ {
LOG(ERROR) << "Failed creating socket " << fId << ", reason: " << zmq_strerror(errno); LOG(ERROR) << "Failed creating socket " << fId << ", reason: " << zmq_strerror(errno);
exit(EXIT_FAILURE); exit(EXIT_FAILURE);
@ -67,7 +68,7 @@ FairMQSocketSHM::FairMQSocketSHM(const string& type, const string& name, const s
if (type == "sub") if (type == "sub")
{ {
if (zmq_setsockopt(fSocket, ZMQ_SUBSCRIBE, NULL, 0) != 0) if (zmq_setsockopt(fSocket, ZMQ_SUBSCRIBE, nullptr, 0) != 0)
{ {
LOG(ERROR) << "Failed setting ZMQ_SUBSCRIBE socket option, reason: " << zmq_strerror(errno); LOG(ERROR) << "Failed setting ZMQ_SUBSCRIBE socket option, reason: " << zmq_strerror(errno);
} }
@ -301,7 +302,7 @@ int64_t FairMQSocketSHM::Receive(vector<FairMQMessagePtr>& msgVec, const int fla
do do
{ {
FairMQMessagePtr part(new FairMQMessageSHM()); FairMQMessagePtr part(new FairMQMessageSHM(fManager));
zmq_msg_t* msgPtr = static_cast<zmq_msg_t*>(part->GetMessage()); zmq_msg_t* msgPtr = static_cast<zmq_msg_t*>(part->GetMessage());
int nbytes = zmq_msg_recv(msgPtr, fSocket, flags); int nbytes = zmq_msg_recv(msgPtr, fSocket, flags);
@ -360,7 +361,7 @@ void FairMQSocketSHM::Close()
{ {
// LOG(DEBUG) << "Closing socket " << fId; // LOG(DEBUG) << "Closing socket " << fId;
if (fSocket == NULL) if (fSocket == nullptr)
{ {
return; return;
} }
@ -370,20 +371,20 @@ void FairMQSocketSHM::Close()
LOG(ERROR) << "Failed closing socket " << fId << ", reason: " << zmq_strerror(errno); LOG(ERROR) << "Failed closing socket " << fId << ", reason: " << zmq_strerror(errno);
} }
fSocket = NULL; fSocket = nullptr;
} }
void FairMQSocketSHM::Interrupt() void FairMQSocketSHM::Interrupt()
{ {
fManager.Interrupt();
FairMQMessageSHM::fInterrupted = true; FairMQMessageSHM::fInterrupted = true;
FairMQUnmanagedRegionSHM::fInterrupted = true;
fInterrupted = true; fInterrupted = true;
} }
void FairMQSocketSHM::Resume() void FairMQSocketSHM::Resume()
{ {
fManager.Resume();
FairMQMessageSHM::fInterrupted = false; FairMQMessageSHM::fInterrupted = false;
FairMQUnmanagedRegionSHM::fInterrupted = true;
fInterrupted = false; fInterrupted = false;
} }

55
fairmq/shmem/FairMQSocketSHM.h
View File

@ -1,8 +1,8 @@
/******************************************************************************** /********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH * * Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* * * *
* This software is distributed under the terms of the * * This software is distributed under the terms of the *
* GNU Lesser General Public Licence version 3 (LGPL) version 3, * * GNU Lesser General Public Licence version 3 (LGPL) version 3, *
* copied verbatim in the file "LICENSE" * * copied verbatim in the file "LICENSE" *
********************************************************************************/ ********************************************************************************/
#ifndef FAIRMQSOCKETSHM_H_ #ifndef FAIRMQSOCKETSHM_H_
@ -14,52 +14,53 @@
#include "FairMQSocket.h" #include "FairMQSocket.h"
#include "FairMQMessage.h" #include "FairMQMessage.h"
#include "FairMQShmManager.h" #include "Manager.h"
class FairMQSocketSHM : public FairMQSocket class FairMQSocketSHM : public FairMQSocket
{ {
public: public:
FairMQSocketSHM(const std::string& type, const std::string& name, const std::string& id = "", void* context = nullptr); FairMQSocketSHM(fair::mq::shmem::Manager& manager, const std::string& type, const std::string& name, const std::string& id = "", void* context = nullptr);
FairMQSocketSHM(const FairMQSocketSHM&) = delete; FairMQSocketSHM(const FairMQSocketSHM&) = delete;
FairMQSocketSHM operator=(const FairMQSocketSHM&) = delete; FairMQSocketSHM operator=(const FairMQSocketSHM&) = delete;
virtual std::string GetId(); std::string GetId() override;
virtual bool Bind(const std::string& address); bool Bind(const std::string& address) override;
virtual void Connect(const std::string& address); void Connect(const std::string& address) override;
virtual int Send(FairMQMessagePtr& msg, const int flags = 0); int Send(FairMQMessagePtr& msg, const int flags = 0) override;
virtual int Receive(FairMQMessagePtr& msg, const int flags = 0); int Receive(FairMQMessagePtr& msg, const int flags = 0) override;
virtual int64_t Send(std::vector<std::unique_ptr<FairMQMessage>>& msgVec, const int flags = 0); int64_t Send(std::vector<std::unique_ptr<FairMQMessage>>& msgVec, const int flags = 0) override;
virtual int64_t Receive(std::vector<std::unique_ptr<FairMQMessage>>& msgVec, const int flags = 0); int64_t Receive(std::vector<std::unique_ptr<FairMQMessage>>& msgVec, const int flags = 0) override;
virtual void* GetSocket() const; void* GetSocket() const override;
virtual int GetSocket(int nothing) const; int GetSocket(int nothing) const override;
virtual void Close(); void Close() override;
virtual void Interrupt(); void Interrupt() override;
virtual void Resume(); void Resume() override;
virtual void SetOption(const std::string& option, const void* value, size_t valueSize); void SetOption(const std::string& option, const void* value, size_t valueSize) override;
virtual void GetOption(const std::string& option, void* value, size_t* valueSize); void GetOption(const std::string& option, void* value, size_t* valueSize) override;
virtual unsigned long GetBytesTx() const; unsigned long GetBytesTx() const override;
virtual unsigned long GetBytesRx() const; unsigned long GetBytesRx() const override;
virtual unsigned long GetMessagesTx() const; unsigned long GetMessagesTx() const override;
virtual unsigned long GetMessagesRx() const; unsigned long GetMessagesRx() const override;
virtual bool SetSendTimeout(const int timeout, const std::string& address, const std::string& method); bool SetSendTimeout(const int timeout, const std::string& address, const std::string& method) override;
virtual int GetSendTimeout() const; int GetSendTimeout() const override;
virtual bool SetReceiveTimeout(const int timeout, const std::string& address, const std::string& method); bool SetReceiveTimeout(const int timeout, const std::string& address, const std::string& method) override;
virtual int GetReceiveTimeout() const; int GetReceiveTimeout() const override;
static int GetConstant(const std::string& constant); static int GetConstant(const std::string& constant);
virtual ~FairMQSocketSHM(); ~FairMQSocketSHM() override;
private: private:
void* fSocket; void* fSocket;
fair::mq::shmem::Manager& fManager;
std::string fId; std::string fId;
std::atomic<unsigned long> fBytesTx; std::atomic<unsigned long> fBytesTx;
std::atomic<unsigned long> fBytesRx; std::atomic<unsigned long> fBytesRx;

58
fairmq/shmem/FairMQTransportFactorySHM.cxx
View File

@ -7,7 +7,6 @@
********************************************************************************/ ********************************************************************************/
#include "FairMQLogger.h" #include "FairMQLogger.h"
#include "FairMQShmManager.h"
#include "FairMQTransportFactorySHM.h" #include "FairMQTransportFactorySHM.h"
#include <zmq.h> #include <zmq.h>
@ -27,9 +26,10 @@
using namespace std; using namespace std;
using namespace fair::mq::shmem; using namespace fair::mq::shmem;
namespace bipc = boost::interprocess;
namespace bfs = boost::filesystem; namespace bfs = boost::filesystem;
namespace bpt = boost::posix_time; namespace bpt = boost::posix_time;
namespace bipc = boost::interprocess;
FairMQ::Transport FairMQTransportFactorySHM::fTransportType = FairMQ::Transport::SHM; FairMQ::Transport FairMQTransportFactorySHM::fTransportType = FairMQ::Transport::SHM;
@ -39,8 +39,10 @@ FairMQTransportFactorySHM::FairMQTransportFactorySHM(const string& id, const Fai
, fHeartbeatSocket(nullptr) , fHeartbeatSocket(nullptr)
, fHeartbeatThread() , fHeartbeatThread()
, fSendHeartbeats(true) , fSendHeartbeats(true)
, fShMutex(bipc::open_or_create, "fairmq_shmem_mutex") , fShMutex(bipc::open_or_create, "fmq_shm_mutex")
, fDeviceCounter(nullptr) , fDeviceCounter(nullptr)
, fSegmentName()
, fManager(nullptr)
{ {
int major, minor, patch; int major, minor, patch;
zmq_version(&major, &minor, &patch); zmq_version(&major, &minor, &patch);
@ -53,15 +55,15 @@ FairMQTransportFactorySHM::FairMQTransportFactorySHM(const string& id, const Fai
LOG(ERROR) << "failed creating context, reason: " << zmq_strerror(errno); LOG(ERROR) << "failed creating context, reason: " << zmq_strerror(errno);
exit(EXIT_FAILURE); exit(EXIT_FAILURE);
} }
int numIoThreads = 1; int numIoThreads = 1;
fSegmentName = "fmq_shm_main";
size_t segmentSize = 2000000000; size_t segmentSize = 2000000000;
string segmentName = "fairmq_shmem_main";
if (config) if (config)
{ {
numIoThreads = config->GetValue<int>("io-threads"); numIoThreads = config->GetValue<int>("io-threads");
fSegmentName = config->GetValue<string>("shm-segment-name");
segmentSize = config->GetValue<size_t>("shm-segment-size"); segmentSize = config->GetValue<size_t>("shm-segment-size");
segmentName = config->GetValue<string>("shm-segment-name");
} }
else else
{ {
@ -79,13 +81,13 @@ FairMQTransportFactorySHM::FairMQTransportFactorySHM(const string& id, const Fai
LOG(ERROR) << "shmem: failed configuring context, reason: " << zmq_strerror(errno); LOG(ERROR) << "shmem: failed configuring context, reason: " << zmq_strerror(errno);
} }
Manager::Instance().InitializeSegment("open_or_create", segmentName, segmentSize); fManager = fair::mq::tools::make_unique<Manager>(fSegmentName, segmentSize);
LOG(DEBUG) << "shmem: created/opened shared memory segment of " << segmentSize << " bytes. Available are " << Manager::Instance().Segment()->get_free_memory() << " bytes."; LOG(DEBUG) << "shmem: created/opened shared memory segment of " << segmentSize << " bytes. Available are " << fManager->Segment().get_free_memory() << " bytes.";
{ {
bipc::scoped_lock<bipc::named_mutex> lock(fShMutex); bipc::scoped_lock<bipc::named_mutex> lock(fShMutex);
fDeviceCounter = Manager::Instance().Segment()->find<DeviceCounter>(bipc::unique_instance).first; fDeviceCounter = fManager->Segment().find<DeviceCounter>(bipc::unique_instance).first;
if (fDeviceCounter) if (fDeviceCounter)
{ {
LOG(DEBUG) << "shmem: device counter found, with value of " << fDeviceCounter->fCount << ". incrementing."; LOG(DEBUG) << "shmem: device counter found, with value of " << fDeviceCounter->fCount << ". incrementing.";
@ -95,7 +97,7 @@ FairMQTransportFactorySHM::FairMQTransportFactorySHM(const string& id, const Fai
else else
{ {
LOG(DEBUG) << "shmem: no device counter found, creating one and initializing with 1"; LOG(DEBUG) << "shmem: no device counter found, creating one and initializing with 1";
fDeviceCounter = Manager::Instance().Segment()->construct<DeviceCounter>(bipc::unique_instance)(1); fDeviceCounter = fManager->Segment().construct<DeviceCounter>(bipc::unique_instance)(1);
LOG(DEBUG) << "shmem: initialized device counter with: " << fDeviceCounter->fCount; LOG(DEBUG) << "shmem: initialized device counter with: " << fDeviceCounter->fCount;
} }
@ -110,7 +112,7 @@ FairMQTransportFactorySHM::FairMQTransportFactorySHM(const string& id, const Fai
// } // }
// else // else
// { // {
// LOG(DEBUG) << "shmem: found shmmonitor in fairmq_shmem_management."; // LOG(DEBUG) << "shmem: found shmmonitor in fmq_shm_management.";
// } // }
// } // }
// catch (std::exception& e) // catch (std::exception& e)
@ -140,7 +142,7 @@ void FairMQTransportFactorySHM::StartMonitor()
do do
{ {
MonitorStatus* monitorStatus = Manager::Instance().ManagementSegment().find<MonitorStatus>(bipc::unique_instance).first; MonitorStatus* monitorStatus = fManager->ManagementSegment().find<MonitorStatus>(bipc::unique_instance).first;
if (monitorStatus) if (monitorStatus)
{ {
LOG(DEBUG) << "shmem: shmmonitor started"; LOG(DEBUG) << "shmem: shmmonitor started";
@ -165,7 +167,7 @@ void FairMQTransportFactorySHM::SendHeartbeats()
{ {
try try
{ {
bipc::message_queue mq(bipc::open_only, "fairmq_shmem_control_queue"); bipc::message_queue mq(bipc::open_only, "fmq_shm_control_queue");
bool heartbeat = true; bool heartbeat = true;
bpt::ptime sndTill = bpt::microsec_clock::universal_time() + bpt::milliseconds(100); bpt::ptime sndTill = bpt::microsec_clock::universal_time() + bpt::milliseconds(100);
if (mq.timed_send(&heartbeat, sizeof(heartbeat), 0, sndTill)) if (mq.timed_send(&heartbeat, sizeof(heartbeat), 0, sndTill))
@ -180,35 +182,35 @@ void FairMQTransportFactorySHM::SendHeartbeats()
catch (bipc::interprocess_exception& ie) catch (bipc::interprocess_exception& ie)
{ {
this_thread::sleep_for(chrono::milliseconds(500)); this_thread::sleep_for(chrono::milliseconds(500));
// LOG(WARN) << "no fairmq_shmem_control_queue found"; // LOG(WARN) << "no fmq_shm_control_queue found";
} }
} }
} }
FairMQMessagePtr FairMQTransportFactorySHM::CreateMessage() const FairMQMessagePtr FairMQTransportFactorySHM::CreateMessage() const
{ {
return unique_ptr<FairMQMessage>(new FairMQMessageSHM()); return unique_ptr<FairMQMessage>(new FairMQMessageSHM(*fManager));
} }
FairMQMessagePtr FairMQTransportFactorySHM::CreateMessage(const size_t size) const FairMQMessagePtr FairMQTransportFactorySHM::CreateMessage(const size_t size) const
{ {
return unique_ptr<FairMQMessage>(new FairMQMessageSHM(size)); return unique_ptr<FairMQMessage>(new FairMQMessageSHM(*fManager, size));
} }
FairMQMessagePtr FairMQTransportFactorySHM::CreateMessage(void* data, const size_t size, fairmq_free_fn* ffn, void* hint) const FairMQMessagePtr FairMQTransportFactorySHM::CreateMessage(void* data, const size_t size, fairmq_free_fn* ffn, void* hint) const
{ {
return unique_ptr<FairMQMessage>(new FairMQMessageSHM(data, size, ffn, hint)); return unique_ptr<FairMQMessage>(new FairMQMessageSHM(*fManager, data, size, ffn, hint));
} }
FairMQMessagePtr FairMQTransportFactorySHM::CreateMessage(FairMQUnmanagedRegionPtr& region, void* data, const size_t size) const FairMQMessagePtr FairMQTransportFactorySHM::CreateMessage(FairMQUnmanagedRegionPtr& region, void* data, const size_t size) const
{ {
return unique_ptr<FairMQMessage>(new FairMQMessageSHM(region, data, size)); return unique_ptr<FairMQMessage>(new FairMQMessageSHM(*fManager, region, data, size));
} }
FairMQSocketPtr FairMQTransportFactorySHM::CreateSocket(const string& type, const string& name) const FairMQSocketPtr FairMQTransportFactorySHM::CreateSocket(const string& type, const string& name) const
{ {
assert(fContext); assert(fContext);
return unique_ptr<FairMQSocket>(new FairMQSocketSHM(type, name, GetId(), fContext)); return unique_ptr<FairMQSocket>(new FairMQSocketSHM(*fManager, type, name, GetId(), fContext));
} }
FairMQPollerPtr FairMQTransportFactorySHM::CreatePoller(const vector<FairMQChannel>& channels) const FairMQPollerPtr FairMQTransportFactorySHM::CreatePoller(const vector<FairMQChannel>& channels) const
@ -231,9 +233,9 @@ FairMQPollerPtr FairMQTransportFactorySHM::CreatePoller(const FairMQSocket& cmdS
return unique_ptr<FairMQPoller>(new FairMQPollerSHM(cmdSocket, dataSocket)); return unique_ptr<FairMQPoller>(new FairMQPollerSHM(cmdSocket, dataSocket));
} }
FairMQUnmanagedRegionPtr FairMQTransportFactorySHM::CreateUnmanagedRegion(const size_t size) const FairMQUnmanagedRegionPtr FairMQTransportFactorySHM::CreateUnmanagedRegion(const size_t size, FairMQRegionCallback callback) const
{ {
return unique_ptr<FairMQUnmanagedRegion>(new FairMQUnmanagedRegionSHM(size)); return unique_ptr<FairMQUnmanagedRegion>(new FairMQUnmanagedRegionSHM(*fManager, size, callback));
} }
FairMQTransportFactorySHM::~FairMQTransportFactorySHM() FairMQTransportFactorySHM::~FairMQTransportFactorySHM()
@ -261,6 +263,8 @@ FairMQTransportFactorySHM::~FairMQTransportFactorySHM()
LOG(ERROR) << "shmem: Terminate(): context not available for shutdown"; LOG(ERROR) << "shmem: Terminate(): context not available for shutdown";
} }
bool lastRemoved = false;
{ // mutex scope { // mutex scope
bipc::scoped_lock<bipc::named_mutex> lock(fShMutex); bipc::scoped_lock<bipc::named_mutex> lock(fShMutex);
@ -268,15 +272,21 @@ FairMQTransportFactorySHM::~FairMQTransportFactorySHM()
if (fDeviceCounter->fCount == 0) if (fDeviceCounter->fCount == 0)
{ {
LOG(DEBUG) << "shmem: last 'fairmq_shmem_main' user, removing segment."; LOG(DEBUG) << "shmem: last " << fSegmentName << " user, removing segment.";
Manager::Instance().Remove(); fManager->RemoveSegment();
lastRemoved = true;
} }
else else
{ {
LOG(DEBUG) << "shmem: other 'fairmq_shmem_main' users present (" << fDeviceCounter->fCount << "), not removing it."; LOG(DEBUG) << "shmem: other " << fSegmentName << " users present (" << fDeviceCounter->fCount << "), not removing it.";
} }
} }
if (lastRemoved)
{
boost::interprocess::named_mutex::remove("fmq_shm_mutex");
}
} }
FairMQ::Transport FairMQTransportFactorySHM::GetType() const FairMQ::Transport FairMQTransportFactorySHM::GetType() const

13
fairmq/shmem/FairMQTransportFactorySHM.h
View File

@ -10,20 +10,21 @@
#define FAIRMQTRANSPORTFACTORYSHM_H_ #define FAIRMQTRANSPORTFACTORYSHM_H_
#include "FairMQTransportFactory.h" #include "FairMQTransportFactory.h"
#include "Manager.h"
#include "FairMQMessageSHM.h" #include "FairMQMessageSHM.h"
#include "FairMQSocketSHM.h" #include "FairMQSocketSHM.h"
#include "FairMQPollerSHM.h" #include "FairMQPollerSHM.h"
#include "FairMQShmCommon.h" #include "Common.h"
#include <options/FairMQProgOptions.h>
#include "FairMQUnmanagedRegionSHM.h" #include "FairMQUnmanagedRegionSHM.h"
#include <options/FairMQProgOptions.h>
#include <boost/interprocess/sync/named_mutex.hpp>
#include <vector> #include <vector>
#include <string> #include <string>
#include <thread> #include <thread>
#include <atomic> #include <atomic>
#include <boost/interprocess/sync/named_mutex.hpp>
class FairMQTransportFactorySHM : public FairMQTransportFactory class FairMQTransportFactorySHM : public FairMQTransportFactory
{ {
public: public:
@ -43,7 +44,7 @@ class FairMQTransportFactorySHM : public FairMQTransportFactory
FairMQPollerPtr CreatePoller(const std::unordered_map<std::string, std::vector<FairMQChannel>>& channelsMap, const std::vector<std::string>& channelList) const override; FairMQPollerPtr CreatePoller(const std::unordered_map<std::string, std::vector<FairMQChannel>>& channelsMap, const std::vector<std::string>& channelList) const override;
FairMQPollerPtr CreatePoller(const FairMQSocket& cmdSocket, const FairMQSocket& dataSocket) const override; FairMQPollerPtr CreatePoller(const FairMQSocket& cmdSocket, const FairMQSocket& dataSocket) const override;
FairMQUnmanagedRegionPtr CreateUnmanagedRegion(const size_t size) const override; FairMQUnmanagedRegionPtr CreateUnmanagedRegion(const size_t size, FairMQRegionCallback callback = nullptr) const override;
FairMQ::Transport GetType() const override; FairMQ::Transport GetType() const override;
@ -60,6 +61,8 @@ class FairMQTransportFactorySHM : public FairMQTransportFactory
std::atomic<bool> fSendHeartbeats; std::atomic<bool> fSendHeartbeats;
boost::interprocess::named_mutex fShMutex; boost::interprocess::named_mutex fShMutex;
fair::mq::shmem::DeviceCounter* fDeviceCounter; fair::mq::shmem::DeviceCounter* fDeviceCounter;
std::string fSegmentName;
std::unique_ptr<fair::mq::shmem::Manager> fManager;
}; };
#endif /* FAIRMQTRANSPORTFACTORYSHM_H_ */ #endif /* FAIRMQTRANSPORTFACTORYSHM_H_ */

52
fairmq/shmem/FairMQUnmanagedRegionSHM.cxx
View File

@ -7,26 +7,21 @@
********************************************************************************/ ********************************************************************************/
#include "FairMQUnmanagedRegionSHM.h" #include "FairMQUnmanagedRegionSHM.h"
#include "FairMQShmManager.h" #include "Common.h"
#include "FairMQShmCommon.h"
using namespace std; using namespace std;
using namespace fair::mq::shmem; using namespace fair::mq::shmem;
namespace bipc = boost::interprocess; namespace bipc = boost::interprocess;
atomic<bool> FairMQUnmanagedRegionSHM::fInterrupted(false); FairMQUnmanagedRegionSHM::FairMQUnmanagedRegionSHM(Manager& manager, const size_t size, FairMQRegionCallback callback)
unordered_map<uint64_t, RemoteRegion> FairMQUnmanagedRegionSHM::fRemoteRegionMap; : fManager(manager)
, fRegion(nullptr)
FairMQUnmanagedRegionSHM::FairMQUnmanagedRegionSHM(const size_t size)
: fRegion(nullptr)
, fRegionId(0) , fRegionId(0)
, fRegionIdStr()
, fRemote(false)
{ {
try try
{ {
RegionCounter* rc = Manager::Instance().ManagementSegment().find<RegionCounter>(bipc::unique_instance).first; RegionCounter* rc = fManager.ManagementSegment().find<RegionCounter>(bipc::unique_instance).first;
if (rc) if (rc)
{ {
LOG(DEBUG) << "shmem: region counter found, with value of " << rc->fCount << ". incrementing."; LOG(DEBUG) << "shmem: region counter found, with value of " << rc->fCount << ". incrementing.";
@ -36,29 +31,13 @@ FairMQUnmanagedRegionSHM::FairMQUnmanagedRegionSHM(const size_t size)
else else
{ {
LOG(DEBUG) << "shmem: no region counter found, creating one and initializing with 1"; LOG(DEBUG) << "shmem: no region counter found, creating one and initializing with 1";
rc = Manager::Instance().ManagementSegment().construct<RegionCounter>(bipc::unique_instance)(1); rc = fManager.ManagementSegment().construct<RegionCounter>(bipc::unique_instance)(1);
LOG(DEBUG) << "shmem: initialized region counter with: " << rc->fCount; LOG(DEBUG) << "shmem: initialized region counter with: " << rc->fCount;
} }
fRegionId = rc->fCount; fRegionId = rc->fCount;
fRegionIdStr = "fairmq_shmem_region_" + to_string(fRegionId);
auto it = fRemoteRegionMap.find(fRegionId); fRegion = fManager.CreateRegion(size, fRegionId, callback);
if (it != fRemoteRegionMap.end())
{
LOG(ERROR) << "Trying to create a region that already exists";
}
else
{
string regionIdStr = "fairmq_shmem_region_" + to_string(fRegionId);
LOG(DEBUG) << "creating region with id " << fRegionId;
auto r = fRemoteRegionMap.emplace(fRegionId, RemoteRegion{regionIdStr, size});
fRegion = &(r.first->second.fRegion);
LOG(DEBUG) << "created region with id " << fRegionId;
}
} }
catch (bipc::interprocess_exception& e) catch (bipc::interprocess_exception& e)
{ {
@ -78,22 +57,7 @@ size_t FairMQUnmanagedRegionSHM::GetSize() const
return fRegion->get_size(); return fRegion->get_size();
} }
bipc::mapped_region* FairMQUnmanagedRegionSHM::GetRemoteRegion(uint64_t regionId)
{
auto it = fRemoteRegionMap.find(regionId);
if (it != fRemoteRegionMap.end())
{
return &(it->second.fRegion);
}
else
{
string regionIdStr = "fairmq_shmem_region_" + to_string(regionId);
auto r = fRemoteRegionMap.emplace(regionId, RemoteRegion{regionIdStr, 0});
return &(r.first->second.fRegion);
}
}
FairMQUnmanagedRegionSHM::~FairMQUnmanagedRegionSHM() FairMQUnmanagedRegionSHM::~FairMQUnmanagedRegionSHM()
{ {
fManager.RemoveRegion(fRegionId);
} }

47
fairmq/shmem/FairMQUnmanagedRegionSHM.h
View File

@ -11,6 +11,7 @@
#include "FairMQUnmanagedRegion.h" #include "FairMQUnmanagedRegion.h"
#include "FairMQLogger.h" #include "FairMQLogger.h"
#include "Manager.h"
#include <boost/interprocess/shared_memory_object.hpp> #include <boost/interprocess/shared_memory_object.hpp>
#include <boost/interprocess/mapped_region.hpp> #include <boost/interprocess/mapped_region.hpp>
@ -21,59 +22,23 @@
#include <string> #include <string>
#include <unordered_map> #include <unordered_map>
struct RemoteRegion // todo: better name?
{
RemoteRegion(std::string regionIdStr, uint64_t size)
: fRegionName(regionIdStr)
, fShmemObject(boost::interprocess::open_or_create, regionIdStr.c_str(), boost::interprocess::read_write)
{
if (size > 0)
{
fShmemObject.truncate(size);
}
fRegion = boost::interprocess::mapped_region(fShmemObject, boost::interprocess::read_write); // TODO: add HUGEPAGES flag here
}
RemoteRegion() = delete;
RemoteRegion(const RemoteRegion& rr) = default;
RemoteRegion(RemoteRegion&& rr) = default;
~RemoteRegion()
{
if (boost::interprocess::shared_memory_object::remove(fRegionName.c_str()))
{
LOG(DEBUG) << "destroyed region " << fRegionName;
}
}
std::string fRegionName;
boost::interprocess::shared_memory_object fShmemObject;
boost::interprocess::mapped_region fRegion;
};
class FairMQUnmanagedRegionSHM : public FairMQUnmanagedRegion class FairMQUnmanagedRegionSHM : public FairMQUnmanagedRegion
{ {
friend class FairMQSocketSHM; friend class FairMQSocketSHM;
friend class FairMQMessageSHM; friend class FairMQMessageSHM;
public: public:
FairMQUnmanagedRegionSHM(const size_t size); FairMQUnmanagedRegionSHM(fair::mq::shmem::Manager& manager, const size_t size, FairMQRegionCallback callback = nullptr);
virtual void* GetData() const override; void* GetData() const override;
virtual size_t GetSize() const override; size_t GetSize() const override;
static boost::interprocess::mapped_region* GetRemoteRegion(uint64_t regionId); ~FairMQUnmanagedRegionSHM() override;
virtual ~FairMQUnmanagedRegionSHM();
private: private:
static std::atomic<bool> fInterrupted; fair::mq::shmem::Manager& fManager;
boost::interprocess::mapped_region* fRegion; boost::interprocess::mapped_region* fRegion;
uint64_t fRegionId; uint64_t fRegionId;
std::string fRegionIdStr;
bool fRemote;
static std::unordered_map<uint64_t, RemoteRegion> fRemoteRegionMap;
}; };
#endif /* FAIRMQUNMANAGEDREGIONSHM_H_ */ #endif /* FAIRMQUNMANAGEDREGIONSHM_H_ */

141
fairmq/shmem/Manager.cxx Normal file
View File

@ -0,0 +1,141 @@
/********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence version 3 (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#include "Manager.h"
#include "Common.h"
namespace fair
{
namespace mq
{
namespace shmem
{
using namespace std;
namespace bipc = boost::interprocess;
Manager::Manager(const string& name, size_t size)
: fName(name)
, fSegment(bipc::open_or_create, fName.c_str(), size)
, fManagementSegment(bipc::open_or_create, "fmq_shm_management", 65536)
, fRegions()
{}
bipc::managed_shared_memory& Manager::Segment()
{
return fSegment;
}
void Manager::Interrupt()
{
}
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;
}
}
}
bipc::mapped_region* Manager::CreateRegion(const size_t size, const uint64_t id, FairMQRegionCallback callback)
{
auto it = fRegions.find(id);
if (it != fRegions.end())
{
LOG(ERROR) << "shmem: Trying to create a region that already exists";
return nullptr;
}
else
{
auto r = fRegions.emplace(id, Region{*this, id, size, false, callback});
r.first->second.StartReceivingAcks();
return &(r.first->second.fRegion);
}
}
bipc::mapped_region* Manager::GetRemoteRegion(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);
if (it != fRegions.end())
{
return &(it->second.fRegion);
}
else
{
try
{
auto r = fRegions.emplace(id, Region{*this, id, 0, true, nullptr});
return &(r.first->second.fRegion);
}
catch (bipc::interprocess_exception& e)
{
// LOG(WARN) << "remote region (" << id << ") no longer exists";
return nullptr;
}
}
}
void Manager::RemoveRegion(const uint64_t id)
{
fRegions.erase(id);
}
bipc::message_queue* Manager::GetRegionQueue(const uint64_t id)
{
try
{
return fRegions.at(id).fQueue.get();
}
catch (out_of_range& oor)
{
return nullptr;
}
}
void Manager::RemoveSegment()
{
if (bipc::shared_memory_object::remove(fName.c_str()))
{
LOG(DEBUG) << "shmem: successfully removed " << fName << " segment after the device has stopped.";
}
else
{
LOG(DEBUG) << "shmem: did not remove " << fName << " segment after the device stopped. Already removed?";
}
if (bipc::shared_memory_object::remove("fmq_shm_management"))
{
LOG(DEBUG) << "shmem: successfully removed \"fmq_shm_management\" segment after the device has stopped.";
}
else
{
LOG(DEBUG) << "shmem: did not remove \"fmq_shm_management\" segment after the device stopped. Already removed?";
}
}
bipc::managed_shared_memory& Manager::ManagementSegment()
{
return fManagementSegment;
}
} // namespace shmem
} // namespace mq
} // namespace fair

81
fairmq/shmem/Manager.h Normal file
View File

@ -0,0 +1,81 @@
/********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence version 3 (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
/**
* FairMQShmManager.h
*
* @since 2016-04-08
* @author A. Rybalchenko
*/
#ifndef FAIR_MQ_SHMEM_MANAGER_H_
#define FAIR_MQ_SHMEM_MANAGER_H_
#include "FairMQLogger.h"
#include "FairMQMessage.h"
#include "fairmq/Tools.h"
#include "Region.h"
#include "Common.h"
#include <boost/interprocess/managed_shared_memory.hpp>
#include <boost/interprocess/ipc/message_queue.hpp>
#include <thread>
#include <queue>
#include <string>
#include <vector>
#include <functional>
#include <unordered_map>
#include <mutex>
#include <condition_variable>
namespace fair
{
namespace mq
{
namespace shmem
{
class Manager
{
friend class Region;
public:
Manager(const std::string& name, size_t size);
Manager() = delete;
Manager(const Manager&) = delete;
Manager operator=(const Manager&) = delete;
boost::interprocess::managed_shared_memory& Segment();
void Interrupt();
void Resume();
boost::interprocess::mapped_region* CreateRegion(const size_t size, const uint64_t id, FairMQRegionCallback callback);
boost::interprocess::mapped_region* GetRemoteRegion(const uint64_t id);
void RemoveRegion(const uint64_t id);
boost::interprocess::message_queue* GetRegionQueue(const uint64_t id);
void RemoveSegment();
boost::interprocess::managed_shared_memory& ManagementSegment();
private:
std::string fName;
boost::interprocess::managed_shared_memory fSegment;
boost::interprocess::managed_shared_memory fManagementSegment;
std::unordered_map<uint64_t, Region> fRegions;
};
} // namespace shmem
} // namespace mq
} // namespace fair
#endif /* FAIR_MQ_SHMEM_MANAGER_H_ */

41
fairmq/shmem/FairMQShmMonitor.cxx → fairmq/shmem/Monitor.cxx
View File

@ -6,8 +6,8 @@
* copied verbatim in the file "LICENSE" * * copied verbatim in the file "LICENSE" *
********************************************************************************/ ********************************************************************************/
#include "FairMQShmMonitor.h" #include "Monitor.h"
#include "FairMQShmCommon.h" #include "Common.h"
#include <boost/interprocess/managed_shared_memory.hpp> #include <boost/interprocess/managed_shared_memory.hpp>
#include <boost/interprocess/containers/vector.hpp> #include <boost/interprocess/containers/vector.hpp>
@ -36,7 +36,7 @@ using StringVector = bipc::vector<String, StringAllocator>;
namespace namespace
{ {
volatile std::sig_atomic_t gSignalStatus; volatile std::sig_atomic_t gSignalStatus = 0;
} }
namespace fair namespace fair
@ -61,7 +61,7 @@ Monitor::Monitor(const string& segmentName, bool selfDestruct, bool interactive,
, fHeartbeatTriggered(false) , fHeartbeatTriggered(false)
, fLastHeartbeat() , fLastHeartbeat()
, fSignalThread() , fSignalThread()
, fManagementSegment(bipc::open_or_create, "fairmq_shmem_management", 65536) , fManagementSegment(bipc::open_or_create, "fmq_shm_management", 65536)
{ {
MonitorStatus* monitorStatus = fManagementSegment.find<MonitorStatus>(bipc::unique_instance).first; MonitorStatus* monitorStatus = fManagementSegment.find<MonitorStatus>(bipc::unique_instance).first;
if (monitorStatus != nullptr) if (monitorStatus != nullptr)
@ -71,7 +71,7 @@ Monitor::Monitor(const string& segmentName, bool selfDestruct, bool interactive,
} }
fManagementSegment.construct<MonitorStatus>(bipc::unique_instance)(); fManagementSegment.construct<MonitorStatus>(bipc::unique_instance)();
CleanupControlQueues(); RemoveQueue("fmq_shm_control_queue");
} }
void Monitor::CatchSignals() void Monitor::CatchSignals()
@ -124,7 +124,7 @@ void Monitor::MonitorHeartbeats()
{ {
try try
{ {
bipc::message_queue mq(bipc::open_or_create, "fairmq_shmem_control_queue", 1000, sizeof(bool)); bipc::message_queue mq(bipc::open_or_create, "fmq_shm_control_queue", 1000, sizeof(bool));
unsigned int priority; unsigned int priority;
bipc::message_queue::size_type recvdSize; bipc::message_queue::size_type recvdSize;
@ -149,7 +149,7 @@ void Monitor::MonitorHeartbeats()
cout << ie.what() << endl; cout << ie.what() << endl;
} }
CleanupControlQueues(); RemoveQueue("fmq_shm_control_queue");
} }
void Monitor::Interactive() void Monitor::Interactive()
@ -339,7 +339,7 @@ void Monitor::Cleanup(const string& segmentName)
{ {
try try
{ {
bipc::managed_shared_memory managementSegment(bipc::open_only, "fairmq_shmem_management"); bipc::managed_shared_memory managementSegment(bipc::open_only, "fmq_shm_management");
RegionCounter* rc = managementSegment.find<RegionCounter>(bipc::unique_instance).first; RegionCounter* rc = managementSegment.find<RegionCounter>(bipc::unique_instance).first;
if (rc) if (rc)
{ {
@ -347,7 +347,8 @@ void Monitor::Cleanup(const string& segmentName)
unsigned int regionCount = rc->fCount; unsigned int regionCount = rc->fCount;
for (unsigned int i = 1; i <= regionCount; ++i) for (unsigned int i = 1; i <= regionCount; ++i)
{ {
RemoveObject("fairmq_shmem_region_" + to_string(i)); RemoveObject("fmq_shm_region_" + to_string(i));
RemoveQueue(std::string("fmq_shm_region_queue_" + std::to_string(i)));
} }
} }
else else
@ -355,39 +356,41 @@ void Monitor::Cleanup(const string& segmentName)
cout << "shmem: no region counter found. no regions to cleanup." << endl; cout << "shmem: no region counter found. no regions to cleanup." << endl;
} }
RemoveObject("fairmq_shmem_management"); RemoveObject("fmq_shm_management");
} }
catch (bipc::interprocess_exception& ie) catch (bipc::interprocess_exception& ie)
{ {
cout << "Did not find \"fairmq_shmem_management\" shared memory segment. No regions to cleanup." << endl; cout << "Did not find \"fmq_shm_management\" shared memory segment. No regions to cleanup." << endl;
} }
RemoveObject(segmentName); RemoveObject(segmentName);
boost::interprocess::named_mutex::remove("fairmq_shmem_mutex"); boost::interprocess::named_mutex::remove("fmq_shm_mutex");
cout << endl;
} }
void Monitor::RemoveObject(const std::string& name) void Monitor::RemoveObject(const std::string& name)
{ {
if (bipc::shared_memory_object::remove(name.c_str())) if (bipc::shared_memory_object::remove(name.c_str()))
{ {
cout << "Successfully removed \"" << name << "\" shared memory segment." << endl; cout << "Successfully removed \"" << name << "\"." << endl;
} }
else else
{ {
cout << "Did not remove \"" << name << "\" shared memory segment. Already removed?" << endl; cout << "Did not remove \"" << name << "\". Already removed?" << endl;
} }
} }
void Monitor::CleanupControlQueues() void Monitor::RemoveQueue(const std::string& name)
{ {
if (bipc::message_queue::remove("fairmq_shmem_control_queue")) if (bipc::message_queue::remove(name.c_str()))
{ {
// cout << "successfully removed control queue" << endl; cout << "Successfully removed \"" << name << "\"." << endl;
} }
else else
{ {
// cout << "could not remove control queue" << endl; cout << "Did not remove \"" << name << "\". Already removed?" << endl;
} }
} }
@ -398,7 +401,7 @@ void Monitor::PrintQueues()
try try
{ {
bipc::managed_shared_memory segment(bipc::open_only, fSegmentName.c_str()); bipc::managed_shared_memory segment(bipc::open_only, fSegmentName.c_str());
StringVector* queues = segment.find<StringVector>("fairmq_shmem_queues").first; StringVector* queues = segment.find<StringVector>("fmq_shm_queues").first;
if (queues) if (queues)
{ {
cout << "found " << queues->size() << " queue(s):" << endl; cout << "found " << queues->size() << " queue(s):" << endl;

4
fairmq/shmem/FairMQShmMonitor.h → fairmq/shmem/Monitor.h
View File

@ -36,7 +36,8 @@ class Monitor
virtual ~Monitor(); virtual ~Monitor();
static void Cleanup(const std::string& segmentName); static void Cleanup(const std::string& segmentName);
static void CleanupControlQueues(); static void RemoveObject(const std::string&);
static void RemoveQueue(const std::string&);
private: private:
void PrintHeader(); void PrintHeader();
@ -46,7 +47,6 @@ class Monitor
void CheckSegment(); void CheckSegment();
void Interactive(); void Interactive();
void SignalMonitor(); void SignalMonitor();
static void RemoveObject(const std::string&);
bool fSelfDestruct; // will self-destruct after the memory has been closed bool fSelfDestruct; // will self-destruct after the memory has been closed
bool fInteractive; // running in interactive mode bool fInteractive; // running in interactive mode

16
fairmq/shmem/README.md
View File

@ -10,16 +10,16 @@ Devices track and cleanup shared memory on shutdown. For more information on the
The shared memory monitor tool, supplied with the shared memory transport can be used to monitor shared memory use and automatically cleanup shared memory in case of device crashes. The shared memory monitor tool, supplied with the shared memory transport can be used to monitor shared memory use and automatically cleanup shared memory in case of device crashes.
With default arguments the monitor will run indefinitely with no output, and clean up shared memory segment if it is open and no heartbeats from devices arrive within a timeout period. It can be further customized with following parameters: With default arguments the monitor will run indefinitely with no output, and clean up shared memory segment if it is open and no heartbeats from devices arrive within a timeout period. It can be further customized with following parameters:
`--segment-name <arg>`: customize the name of the shared memory segment (default is "fairmq_shmem_main"). `--segment-name <arg>`: customize the name of the shared memory segment (default is "fmq_shm_main").
`--cleanup`: start monitor, perform cleanup of the memory and quit. `--cleanup`: start monitor, perform cleanup of the memory and quit.
`--self-destruct`: run until the memory segment is closed (either naturally via cleanup performed by devices or in case of a crash (no heartbeats within timeout)). `--self-destruct`: run until the memory segment is closed (either naturally via cleanup performed by devices or in case of a crash (no heartbeats within timeout)).
`--interactive`: run interactively, with detailed segment details and user input for various shmem operations. `--interactive`: run interactively, with detailed segment details and user input for various shmem operations.
`--timeout <arg>`: specifiy the timeout for the heartbeats from shmem transports in milliseconds (default 5000). `--timeout <arg>`: specifiy the timeout for the heartbeats from shmem transports in milliseconds (default 5000).
The options can be combined, with the exception of `--cleanup` option, which will invoke the described behaviour independent of other options. The options can be combined, with the exception of `--cleanup` option, which will invoke the described behaviour independent of other options.
Without the `--self-destruct` option, the monitor will run continously, moitoring (and cleaning up if needed) consecutive topologies. Without the `--self-destruct` option, the monitor will run continuously, moitoring (and cleaning up if needed) consecutive topologies.
Possible further implementation would be to run the monitor with `--self-destruct` with each topology. Possible further implementation would be to run the monitor with `--self-destruct` with each topology.
@ -27,7 +27,9 @@ The FairMQShmMonitor class can also be used independently from the supplied exec
FairMQ Shared Memory currently uses following names to register shared memory on the system: FairMQ Shared Memory currently uses following names to register shared memory on the system:
`fairmq_shmem_main` - main segment name, used for user data (this name can be overridden via `--shm-segment-name`). `fmq_shm_main` - main segment name, used for user data (this name can be overridden via `--shm-segment-name`).
`fairmq_shmem_management` - management segment name, used for storing management data. `fmq_shm_management` - management segment name, used for storing management data.
`fairmq_shmem_control_queue` - message queue for communicating between shm transport and shm monitor (exists independent of above segments). `fmq_shm_control_queue` - message queue for communicating between shm transport and shm monitor (exists independent of above segments).
`fairmq_shmem_mutex` - boost::interprocess::named_mutex for management purposes (exists independent of above segments). `fmq_shm_mutex` - boost::interprocess::named_mutex for management purposes (exists independent of above segments).
`fmq_shm_region_<index>` - names of unmanaged regions.
`fmq_shm_region_queue_<index>` - names of queues for the unmanaged regions.

121
fairmq/shmem/Region.cxx Normal file
View File

@ -0,0 +1,121 @@
/********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence version 3 (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#include "Region.h"
#include "Common.h"
#include "Manager.h"
#include <boost/thread/thread_time.hpp>
#include <chrono>
namespace fair
{
namespace mq
{
namespace shmem
{
using namespace std;
namespace bipc = boost::interprocess;
namespace bpt = boost::posix_time;
Region::Region(Manager& manager, uint64_t id, uint64_t size, bool remote, FairMQRegionCallback callback)
: fManager(manager)
, fRemote(remote)
, fStop(false)
, fName("fmq_shm_region_" + to_string(id))
, fQueueName("fmq_shm_region_queue_" + to_string(id))
, fShmemObject()
, fQueue(nullptr)
, fWorker()
, fCallback(callback)
{
if (fRemote)
{
fShmemObject = bipc::shared_memory_object(bipc::open_only, fName.c_str(), bipc::read_write);
LOG(DEBUG) << "shmem: located remote region: " << fName;
fQueue = fair::mq::tools::make_unique<bipc::message_queue>(bipc::open_only, fQueueName.c_str());
LOG(DEBUG) << "shmem: located remote region queue: " << fQueueName;
}
else
{
fShmemObject = bipc::shared_memory_object(bipc::create_only, fName.c_str(), bipc::read_write);
LOG(DEBUG) << "shmem: created region: " << fName;
fShmemObject.truncate(size);
fQueue = fair::mq::tools::make_unique<bipc::message_queue>(bipc::create_only, fQueueName.c_str(), 10000, sizeof(RegionBlock));
LOG(DEBUG) << "shmem: created region queue: " << fQueueName;
}
fRegion = bipc::mapped_region(fShmemObject, bipc::read_write); // TODO: add HUGEPAGES flag here
// fRegion = bipc::mapped_region(fShmemObject, bipc::read_write, 0, 0, 0, MAP_HUGETLB | MAP_HUGE_1GB);
}
void Region::StartReceivingAcks()
{
fWorker = std::thread(&Region::ReceiveAcks, this);
}
void Region::ReceiveAcks()
{
unsigned int priority;
bipc::message_queue::size_type recvdSize;
while (!fStop) // end thread condition (should exist until region is destroyed)
{
auto rcvTill = bpt::microsec_clock::universal_time() + bpt::milliseconds(200);
RegionBlock block;
if (fQueue->timed_receive(&block, sizeof(RegionBlock), recvdSize, priority, rcvTill))
{
// LOG(DEBUG) << "received: " << block.fHandle << " " << block.fSize << " " << block.fMessageId;
if (fCallback)
{
fCallback(reinterpret_cast<char*>(fRegion.get_address()) + block.fHandle, block.fSize);
}
}
else
{
// LOG(DEBUG) << "queue " << fQueueName << " timeout!";
}
} // while !fStop
LOG(DEBUG) << "worker for " << fName << " leaving.";
}
Region::~Region()
{
if (!fRemote)
{
fStop = true;
if (fWorker.joinable())
{
fWorker.join();
}
if (bipc::shared_memory_object::remove(fName.c_str()))
{
LOG(DEBUG) << "shmem: destroyed region " << fName;
}
if (bipc::message_queue::remove(fQueueName.c_str()))
{
LOG(DEBUG) << "shmem: removed region queue " << fName;
}
}
else
{
// LOG(DEBUG) << "shmem: region '" << fName << "' is remote, no cleanup necessary.";
LOG(DEBUG) << "shmem: region queue '" << fQueueName << "' is remote, no cleanup necessary";
}
}
} // namespace shmem
} // namespace mq
} // namespace fair

68
fairmq/shmem/Region.h Normal file
View File

@ -0,0 +1,68 @@
/********************************************************************************
* Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence version 3 (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
/**
* FairMQShmManager.h
*
* @since 2016-04-08
* @author A. Rybalchenko
*/
#ifndef FAIR_MQ_SHMEM_REGION_H_
#define FAIR_MQ_SHMEM_REGION_H_
#include "FairMQLogger.h"
#include "FairMQUnmanagedRegion.h"
#include "fairmq/Tools.h"
#include <boost/interprocess/managed_shared_memory.hpp>
#include <boost/interprocess/ipc/message_queue.hpp>
#include <thread>
#include <queue>
#include <unordered_map>
namespace fair
{
namespace mq
{
namespace shmem
{
class Manager;
struct Region
{
Region(Manager& manager, uint64_t id, uint64_t size, bool remote, FairMQRegionCallback callback = nullptr);
Region() = delete;
Region(const Region&) = default;
Region(Region&&) = default;
void StartReceivingAcks();
void ReceiveAcks();
~Region();
Manager& fManager;
bool fRemote;
bool fStop;
std::string fName;
std::string fQueueName;
boost::interprocess::shared_memory_object fShmemObject;
boost::interprocess::mapped_region fRegion;
std::unique_ptr<boost::interprocess::message_queue> fQueue;
std::thread fWorker;
FairMQRegionCallback fCallback;
};
} // namespace shmem
} // namespace mq
} // namespace fair
#endif /* FAIR_MQ_SHMEM_REGION_H_ */

6
fairmq/shmem/runFairMQShmMonitor.cxx → fairmq/shmem/runMonitor.cxx
View File

@ -5,7 +5,7 @@
* GNU Lesser General Public Licence version 3 (LGPL) version 3, * * GNU Lesser General Public Licence version 3 (LGPL) version 3, *
* copied verbatim in the file "LICENSE" * * copied verbatim in the file "LICENSE" *
********************************************************************************/ ********************************************************************************/
#include "FairMQShmMonitor.h" #include "Monitor.h"
#include <boost/program_options.hpp> #include <boost/program_options.hpp>
@ -27,7 +27,7 @@ int main(int argc, char** argv)
options_description desc("Options"); options_description desc("Options");
desc.add_options() desc.add_options()
("segment-name", value<string>(&segmentName)->default_value("fairmq_shmem_main"), "Name of the shared memory segment") ("segment-name", value<string>(&segmentName)->default_value("fmq_shm_main"), "Name of the shared memory segment")
("cleanup", value<bool>(&cleanup)->implicit_value(true), "Perform cleanup and quit") ("cleanup", value<bool>(&cleanup)->implicit_value(true), "Perform cleanup and quit")
("self-destruct", value<bool>(&selfDestruct)->implicit_value(true), "Quit after first closing of the memory") ("self-destruct", value<bool>(&selfDestruct)->implicit_value(true), "Quit after first closing of the memory")
("interactive", value<bool>(&interactive)->implicit_value(true), "Interactive run") ("interactive", value<bool>(&interactive)->implicit_value(true), "Interactive run")
@ -49,7 +49,7 @@ int main(int argc, char** argv)
{ {
cout << "Cleaning up \"" << segmentName << "\"..." << endl; cout << "Cleaning up \"" << segmentName << "\"..." << endl;
fair::mq::shmem::Monitor::Cleanup(segmentName); fair::mq::shmem::Monitor::Cleanup(segmentName);
fair::mq::shmem::Monitor::CleanupControlQueues(); fair::mq::shmem::Monitor::RemoveQueue("fmq_shm_control_queue");
return 0; return 0;
} }

11
fairmq/zeromq/FairMQMessageZMQ.cxx
View File

@ -17,10 +17,10 @@
#include "FairMQMessageZMQ.h" #include "FairMQMessageZMQ.h"
#include "FairMQLogger.h" #include "FairMQLogger.h"
#include "FairMQUnmanagedRegionZMQ.h"
using namespace std; using namespace std;
string FairMQMessageZMQ::fDeviceID = string();
FairMQ::Transport FairMQMessageZMQ::fTransportType = FairMQ::Transport::ZMQ; FairMQ::Transport FairMQMessageZMQ::fTransportType = FairMQ::Transport::ZMQ;
FairMQMessageZMQ::FairMQMessageZMQ() FairMQMessageZMQ::FairMQMessageZMQ()
@ -50,7 +50,7 @@ FairMQMessageZMQ::FairMQMessageZMQ(void* data, const size_t size, fairmq_free_fn
} }
} }
FairMQMessageZMQ::FairMQMessageZMQ(FairMQUnmanagedRegionPtr& /*region*/, void* data, const size_t size) FairMQMessageZMQ::FairMQMessageZMQ(FairMQUnmanagedRegionPtr& region, void* data, const size_t size)
: fMessage() : fMessage()
{ {
// FIXME: make this zero-copy: // FIXME: make this zero-copy:
@ -62,6 +62,8 @@ FairMQMessageZMQ::FairMQMessageZMQ(FairMQUnmanagedRegionPtr& /*region*/, void* d
} }
memcpy(zmq_msg_data(&fMessage), data, size); memcpy(zmq_msg_data(&fMessage), data, size);
// call region callback
static_cast<FairMQUnmanagedRegionZMQ*>(region.get())->fCallback(data, size);
// if (zmq_msg_init_data(&fMessage, data, size, [](void*, void*){}, nullptr) != 0) // if (zmq_msg_init_data(&fMessage, data, size, [](void*, void*){}, nullptr) != 0)
// { // {
@ -116,11 +118,6 @@ void FairMQMessageZMQ::SetMessage(void*, const size_t)
// dummy method to comply with the interface. functionality not allowed in zeromq. // dummy method to comply with the interface. functionality not allowed in zeromq.
} }
void FairMQMessageZMQ::SetDeviceId(const string& deviceId)
{
fDeviceID = deviceId;
}
FairMQ::Transport FairMQMessageZMQ::GetType() const FairMQ::Transport FairMQMessageZMQ::GetType() const
{ {
return fTransportType; return fTransportType;

3
fairmq/zeromq/FairMQMessageZMQ.h
View File

@ -41,8 +41,6 @@ class FairMQMessageZMQ : public FairMQMessage
virtual void SetMessage(void* data, const size_t size); virtual void SetMessage(void* data, const size_t size);
virtual void SetDeviceId(const std::string& deviceId);
virtual FairMQ::Transport GetType() const; virtual FairMQ::Transport GetType() const;
virtual void Copy(const std::unique_ptr<FairMQMessage>& msg); virtual void Copy(const std::unique_ptr<FairMQMessage>& msg);
@ -53,7 +51,6 @@ class FairMQMessageZMQ : public FairMQMessage
private: private:
zmq_msg_t fMessage; zmq_msg_t fMessage;
static std::string fDeviceID;
static FairMQ::Transport fTransportType; static FairMQ::Transport fTransportType;
}; };

4
fairmq/zeromq/FairMQTransportFactoryZMQ.cxx
View File

@ -95,9 +95,9 @@ FairMQPollerPtr FairMQTransportFactoryZMQ::CreatePoller(const FairMQSocket& cmdS
return unique_ptr<FairMQPoller>(new FairMQPollerZMQ(cmdSocket, dataSocket)); return unique_ptr<FairMQPoller>(new FairMQPollerZMQ(cmdSocket, dataSocket));
} }
FairMQUnmanagedRegionPtr FairMQTransportFactoryZMQ::CreateUnmanagedRegion(const size_t size) const FairMQUnmanagedRegionPtr FairMQTransportFactoryZMQ::CreateUnmanagedRegion(const size_t size, FairMQRegionCallback callback) const
{ {
return unique_ptr<FairMQUnmanagedRegion>(new FairMQUnmanagedRegionZMQ(size)); return unique_ptr<FairMQUnmanagedRegion>(new FairMQUnmanagedRegionZMQ(size, callback));
} }
FairMQ::Transport FairMQTransportFactoryZMQ::GetType() const FairMQ::Transport FairMQTransportFactoryZMQ::GetType() const

2
fairmq/zeromq/FairMQTransportFactoryZMQ.h
View File

@ -46,7 +46,7 @@ class FairMQTransportFactoryZMQ : public FairMQTransportFactory
FairMQPollerPtr CreatePoller(const std::unordered_map<std::string, std::vector<FairMQChannel>>& channelsMap, const std::vector<std::string>& channelList) const override; FairMQPollerPtr CreatePoller(const std::unordered_map<std::string, std::vector<FairMQChannel>>& channelsMap, const std::vector<std::string>& channelList) const override;
FairMQPollerPtr CreatePoller(const FairMQSocket& cmdSocket, const FairMQSocket& dataSocket) const override; FairMQPollerPtr CreatePoller(const FairMQSocket& cmdSocket, const FairMQSocket& dataSocket) const override;
FairMQUnmanagedRegionPtr CreateUnmanagedRegion(const size_t size) const override; FairMQUnmanagedRegionPtr CreateUnmanagedRegion(const size_t size, FairMQRegionCallback callback) const override;
FairMQ::Transport GetType() const override; FairMQ::Transport GetType() const override;
private: private:

3
fairmq/zeromq/FairMQUnmanagedRegionZMQ.cxx
View File

@ -11,9 +11,10 @@
using namespace std; using namespace std;
FairMQUnmanagedRegionZMQ::FairMQUnmanagedRegionZMQ(const size_t size) FairMQUnmanagedRegionZMQ::FairMQUnmanagedRegionZMQ(const size_t size, FairMQRegionCallback callback)
: fBuffer(malloc(size)) : fBuffer(malloc(size))
, fSize(size) , fSize(size)
, fCallback(callback)
{ {
} }

6
fairmq/zeromq/FairMQUnmanagedRegionZMQ.h
View File

@ -15,10 +15,11 @@
class FairMQUnmanagedRegionZMQ : public FairMQUnmanagedRegion class FairMQUnmanagedRegionZMQ : public FairMQUnmanagedRegion
{ {
friend class FairMQSocketSHM; friend class FairMQSocketZMQ;
friend class FairMQMessageZMQ;
public: public:
FairMQUnmanagedRegionZMQ(const size_t size); FairMQUnmanagedRegionZMQ(const size_t size, FairMQRegionCallback callback);
FairMQUnmanagedRegionZMQ(const FairMQUnmanagedRegionZMQ&) = delete; FairMQUnmanagedRegionZMQ(const FairMQUnmanagedRegionZMQ&) = delete;
FairMQUnmanagedRegionZMQ operator=(const FairMQUnmanagedRegionZMQ&) = delete; FairMQUnmanagedRegionZMQ operator=(const FairMQUnmanagedRegionZMQ&) = delete;
@ -30,6 +31,7 @@ class FairMQUnmanagedRegionZMQ : public FairMQUnmanagedRegion
private: private:
void* fBuffer; void* fBuffer;
size_t fSize; size_t fSize;
FairMQRegionCallback fCallback;
}; };
#endif /* FAIRMQUNMANAGEDREGIONZMQ_H_ */ #endif /* FAIRMQUNMANAGEDREGIONZMQ_H_ */