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

Equip FairMQMessage with pointer to factory (at creation)

Browse Source 
the patch seems big but most of it is just propagating the new notion of
constness of the factory - since it keeps track of created messages with
the internal allocator it no longer is const
This commit is contained in:
mkrzewic 2018-10-18 13:15:04 +02:00 committed by Dennis Klein
parent 310b9647b5
commit d40bbfe208
30 changed files with 195 additions and 177 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
fairmq/FairMQChannel.cxx
View File

@ -689,49 +689,49 @@ void FairMQChannel::ResetChannel()
// TODO: implement channel resetting
}
int FairMQChannel::Send(unique_ptr<FairMQMessage>& msg, int sndTimeoutInMs) const
int FairMQChannel::Send(unique_ptr<FairMQMessage>& msg, int sndTimeoutInMs)
{
CheckSendCompatibility(msg);
return fSocket->Send(msg, sndTimeoutInMs);
}
int FairMQChannel::Receive(unique_ptr<FairMQMessage>& msg, int rcvTimeoutInMs) const
int FairMQChannel::Receive(unique_ptr<FairMQMessage>& msg, int rcvTimeoutInMs)
{
CheckReceiveCompatibility(msg);
return fSocket->Receive(msg, rcvTimeoutInMs);
}
int FairMQChannel::SendAsync(unique_ptr<FairMQMessage>& msg) const
int FairMQChannel::SendAsync(unique_ptr<FairMQMessage>& msg)
{
CheckSendCompatibility(msg);
return fSocket->Send(msg, 0);
}
int FairMQChannel::ReceiveAsync(unique_ptr<FairMQMessage>& msg) const
int FairMQChannel::ReceiveAsync(unique_ptr<FairMQMessage>& msg)
{
CheckReceiveCompatibility(msg);
return fSocket->Receive(msg, 0);
}
int64_t FairMQChannel::Send(vector<unique_ptr<FairMQMessage>>& msgVec, int sndTimeoutInMs) const
int64_t FairMQChannel::Send(vector<unique_ptr<FairMQMessage>>& msgVec, int sndTimeoutInMs)
{
CheckSendCompatibility(msgVec);
return fSocket->Send(msgVec, sndTimeoutInMs);
}
int64_t FairMQChannel::Receive(vector<unique_ptr<FairMQMessage>>& msgVec, int rcvTimeoutInMs) const
int64_t FairMQChannel::Receive(vector<unique_ptr<FairMQMessage>>& msgVec, int rcvTimeoutInMs)
{
CheckReceiveCompatibility(msgVec);
return fSocket->Receive(msgVec, rcvTimeoutInMs);
}
int64_t FairMQChannel::SendAsync(vector<unique_ptr<FairMQMessage>>& msgVec) const
int64_t FairMQChannel::SendAsync(vector<unique_ptr<FairMQMessage>>& msgVec)
{
CheckSendCompatibility(msgVec);
return fSocket->Send(msgVec, 0);
}
int64_t FairMQChannel::ReceiveAsync(vector<unique_ptr<FairMQMessage>>& msgVec) const
int64_t FairMQChannel::ReceiveAsync(vector<unique_ptr<FairMQMessage>>& msgVec)
{
CheckReceiveCompatibility(msgVec);
return fSocket->Receive(msgVec, 0);
@ -761,7 +761,7 @@ unsigned long FairMQChannel::GetMessagesRx() const
return fSocket->GetMessagesRx();
}
void FairMQChannel::CheckSendCompatibility(FairMQMessagePtr& msg) const
void FairMQChannel::CheckSendCompatibility(FairMQMessagePtr& msg)
{
if (fTransportType != msg->GetType())
{
@ -776,7 +776,7 @@ void FairMQChannel::CheckSendCompatibility(FairMQMessagePtr& msg) const
}
}
void FairMQChannel::CheckSendCompatibility(vector<FairMQMessagePtr>& msgVec) const
void FairMQChannel::CheckSendCompatibility(vector<FairMQMessagePtr>& msgVec)
{
for (auto& msg : msgVec)
{
@ -794,7 +794,7 @@ void FairMQChannel::CheckSendCompatibility(vector<FairMQMessagePtr>& msgVec) con
}
}
void FairMQChannel::CheckReceiveCompatibility(FairMQMessagePtr& msg) const
void FairMQChannel::CheckReceiveCompatibility(FairMQMessagePtr& msg)
{
if (fTransportType != msg->GetType())
{
@ -804,7 +804,7 @@ void FairMQChannel::CheckReceiveCompatibility(FairMQMessagePtr& msg) const
}
}
void FairMQChannel::CheckReceiveCompatibility(vector<FairMQMessagePtr>& msgVec) const
void FairMQChannel::CheckReceiveCompatibility(vector<FairMQMessagePtr>& msgVec)
{
for (auto& msg : msgVec)
{

40
fairmq/FairMQChannel.h
View File

@ -178,37 +178,37 @@ class FairMQChannel
/// @param msg Constant reference of unique_ptr to a FairMQMessage
/// @param sndTimeoutInMs send timeout in ms. -1 will wait forever (or until interrupt (e.g. via state change)), 0 will not wait (return immediately if cannot send)
/// @return Number of bytes that have been queued. -2 If queueing was not possible or timed out. -1 if there was an error.
int Send(FairMQMessagePtr& msg, int sndTimeoutInMs = -1) const;
int Send(FairMQMessagePtr& msg, int sndTimeoutInMs = -1);
/// Receives a message from the socket queue.
/// @param msg Constant reference of unique_ptr to a FairMQMessage
/// @param rcvTimeoutInMs receive timeout in ms. -1 will wait forever (or until interrupt (e.g. via state change)), 0 will not wait (return immediately if cannot receive)
/// @return Number of bytes that have been received. -2 if reading from the queue was not possible or timed out. -1 if there was an error.
int Receive(FairMQMessagePtr& msg, int rcvTimeoutInMs = -1) const;
int Receive(FairMQMessagePtr& msg, int rcvTimeoutInMs = -1);
int SendAsync(FairMQMessagePtr& msg) const __attribute__((deprecated("For non-blocking Send, use timeout version with timeout of 0: Send(msg, timeout);")));
int ReceiveAsync(FairMQMessagePtr& msg) const __attribute__((deprecated("For non-blocking Receive, use timeout version with timeout of 0: Receive(msg, timeout);")));
int SendAsync(FairMQMessagePtr& msg) __attribute__((deprecated("For non-blocking Send, use timeout version with timeout of 0: Send(msg, timeout);")));
int ReceiveAsync(FairMQMessagePtr& msg) __attribute__((deprecated("For non-blocking Receive, use timeout version with timeout of 0: Receive(msg, timeout);")));
/// Send a vector of messages
/// @param msgVec message vector reference
/// @param sndTimeoutInMs send timeout in ms. -1 will wait forever (or until interrupt (e.g. via state change)), 0 will not wait (return immediately if cannot send)
/// @return Number of bytes that have been queued. -2 If queueing was not possible or timed out. -1 if there was an error.
int64_t Send(std::vector<FairMQMessagePtr>& msgVec, int sndTimeoutInMs = -1) const;
int64_t Send(std::vector<FairMQMessagePtr>& msgVec, int sndTimeoutInMs = -1);
/// Receive a vector of messages
/// @param msgVec message vector reference
/// @param rcvTimeoutInMs receive timeout in ms. -1 will wait forever (or until interrupt (e.g. via state change)), 0 will not wait (return immediately if cannot receive)
/// @return Number of bytes that have been received. -2 if reading from the queue was not possible or timed out. -1 if there was an error.
int64_t Receive(std::vector<FairMQMessagePtr>& msgVec, int rcvTimeoutInMs = -1) const;
int64_t Receive(std::vector<FairMQMessagePtr>& msgVec, int rcvTimeoutInMs = -1);
int64_t SendAsync(std::vector<FairMQMessagePtr>& msgVec) const __attribute__((deprecated("For non-blocking Send, use timeout version with timeout of 0: Send(msgVec, timeout);")));
int64_t ReceiveAsync(std::vector<FairMQMessagePtr>& msgVec) const __attribute__((deprecated("For non-blocking Receive, use timeout version with timeout of 0: Receive(msgVec, timeout);")));
int64_t SendAsync(std::vector<FairMQMessagePtr>& msgVec) __attribute__((deprecated("For non-blocking Send, use timeout version with timeout of 0: Send(msgVec, timeout);")));
int64_t ReceiveAsync(std::vector<FairMQMessagePtr>& msgVec) __attribute__((deprecated("For non-blocking Receive, use timeout version with timeout of 0: Receive(msgVec, timeout);")));
/// Send FairMQParts
/// @param parts FairMQParts reference
/// @param sndTimeoutInMs send timeout in ms. -1 will wait forever (or until interrupt (e.g. via state change)), 0 will not wait (return immediately if cannot send)
/// @return Number of bytes that have been queued. -2 If queueing was not possible or timed out. -1 if there was an error.
int64_t Send(FairMQParts& parts, int sndTimeoutInMs = -1) const
int64_t Send(FairMQParts& parts, int sndTimeoutInMs = -1)
{
return Send(parts.fParts, sndTimeoutInMs);
}
@ -217,17 +217,17 @@ class FairMQChannel
/// @param parts FairMQParts reference
/// @param rcvTimeoutInMs receive timeout in ms. -1 will wait forever (or until interrupt (e.g. via state change)), 0 will not wait (return immediately if cannot receive)
/// @return Number of bytes that have been received. -2 if reading from the queue was not possible or timed out. -1 if there was an error.
int64_t Receive(FairMQParts& parts, int rcvTimeoutInMs = -1) const
int64_t Receive(FairMQParts& parts, int rcvTimeoutInMs = -1)
{
return Receive(parts.fParts, rcvTimeoutInMs);
}
int64_t SendAsync(FairMQParts& parts) const __attribute__((deprecated("For non-blocking Send, use timeout version with timeout of 0: Send(parts, timeout);")))
int64_t SendAsync(FairMQParts& parts) __attribute__((deprecated("For non-blocking Send, use timeout version with timeout of 0: Send(parts, timeout);")))
{
return Send(parts.fParts, 0);
}
int64_t ReceiveAsync(FairMQParts& parts) const __attribute__((deprecated("For non-blocking Receive, use timeout version with timeout of 0: Receive(parts, timeout);")))
int64_t ReceiveAsync(FairMQParts& parts) __attribute__((deprecated("For non-blocking Receive, use timeout version with timeout of 0: Receive(parts, timeout);")))
{
return Receive(parts.fParts, 0);
}
@ -237,25 +237,25 @@ class FairMQChannel
unsigned long GetMessagesTx() const;
unsigned long GetMessagesRx() const;
auto Transport() const -> const FairMQTransportFactory*
auto Transport() -> FairMQTransportFactory*
{
return fTransportFactory.get();
};
template<typename... Args>
FairMQMessagePtr NewMessage(Args&&... args) const
FairMQMessagePtr NewMessage(Args&&... args)
{
return Transport()->CreateMessage(std::forward<Args>(args)...);
}
template<typename T>
FairMQMessagePtr NewSimpleMessage(const T& data) const
FairMQMessagePtr NewSimpleMessage(const T& data)
{
return Transport()->NewSimpleMessage(data);
}
template<typename T>
FairMQMessagePtr NewStaticMessage(const T& data) const
FairMQMessagePtr NewStaticMessage(const T& data)
{
return Transport()->NewStaticMessage(data);
}
@ -279,10 +279,10 @@ class FairMQChannel
std::shared_ptr<FairMQTransportFactory> fTransportFactory;
void CheckSendCompatibility(FairMQMessagePtr& msg) const;
void CheckSendCompatibility(std::vector<FairMQMessagePtr>& msgVec) const;
void CheckReceiveCompatibility(FairMQMessagePtr& msg) const;
void CheckReceiveCompatibility(std::vector<FairMQMessagePtr>& msgVec) const;
void CheckSendCompatibility(FairMQMessagePtr& msg);
void CheckSendCompatibility(std::vector<FairMQMessagePtr>& msgVec);
void CheckReceiveCompatibility(FairMQMessagePtr& msg);
void CheckReceiveCompatibility(std::vector<FairMQMessagePtr>& msgVec);
void InitTransport(std::shared_ptr<FairMQTransportFactory> factory);

4
fairmq/FairMQDevice.cxx
View File

@ -725,7 +725,7 @@ void FairMQDevice::PollForTransport(const FairMQTransportFactory* factory, const
}
}
bool FairMQDevice::HandleMsgInput(const string& chName, const InputMsgCallback& callback, int i) const
bool FairMQDevice::HandleMsgInput(const string& chName, const InputMsgCallback& callback, int i)
{
unique_ptr<FairMQMessage> input(fChannels.at(chName).at(i).fTransportFactory->CreateMessage());
@ -739,7 +739,7 @@ bool FairMQDevice::HandleMsgInput(const string& chName, const InputMsgCallback&
}
}
bool FairMQDevice::HandleMultipartInput(const string& chName, const InputMultipartCallback& callback, int i) const
bool FairMQDevice::HandleMultipartInput(const string& chName, const InputMultipartCallback& callback, int i)
{
FairMQParts input;

36
fairmq/FairMQDevice.h
View File

@ -102,7 +102,7 @@ class FairMQDevice : public FairMQStateMachine
/// @param i channel index
/// @param sndTimeoutInMs send timeout in ms, -1 will wait forever (or until interrupt (e.g. via state change)), 0 will not wait (return immediately if cannot send)
/// @return Number of bytes that have been queued. -2 If queueing was not possible or timed out. -1 if there was an error.
int Send(FairMQMessagePtr& msg, const std::string& chan, const int i = 0, int sndTimeoutInMs = -1) const
int Send(FairMQMessagePtr& msg, const std::string& chan, const int i = 0, int sndTimeoutInMs = -1)
{
return fChannels.at(chan).at(i).Send(msg, sndTimeoutInMs);
}
@ -113,16 +113,16 @@ class FairMQDevice : public FairMQStateMachine
/// @param i channel index
/// @param rcvTimeoutInMs receive timeout in ms, -1 will wait forever (or until interrupt (e.g. via state change)), 0 will not wait (return immediately if cannot receive)
/// @return Number of bytes that have been received. -2 if reading from the queue was not possible or timed out. -1 if there was an error.
int Receive(FairMQMessagePtr& msg, const std::string& chan, const int i = 0, int rcvTimeoutInMs = -1) const
int Receive(FairMQMessagePtr& msg, const std::string& chan, const int i = 0, int rcvTimeoutInMs = -1)
{
return fChannels.at(chan).at(i).Receive(msg, rcvTimeoutInMs);
}
int SendAsync(FairMQMessagePtr& msg, const std::string& chan, const int i = 0) const __attribute__((deprecated("For non-blocking Send, use timeout version with timeout of 0: Send(msg, \"channelA\", subchannelIndex, timeout);")))
int SendAsync(FairMQMessagePtr& msg, const std::string& chan, const int i = 0) __attribute__((deprecated("For non-blocking Send, use timeout version with timeout of 0: Send(msg, \"channelA\", subchannelIndex, timeout);")))
{
return fChannels.at(chan).at(i).Send(msg, 0);
}
int ReceiveAsync(FairMQMessagePtr& msg, const std::string& chan, const int i = 0) const __attribute__((deprecated("For non-blocking Receive, use timeout version with timeout of 0: Receive(msg, \"channelA\", subchannelIndex, timeout);")))
int ReceiveAsync(FairMQMessagePtr& msg, const std::string& chan, const int i = 0) __attribute__((deprecated("For non-blocking Receive, use timeout version with timeout of 0: Receive(msg, \"channelA\", subchannelIndex, timeout);")))
{
return fChannels.at(chan).at(i).Receive(msg, 0);
}
@ -133,7 +133,7 @@ class FairMQDevice : public FairMQStateMachine
/// @param i channel index
/// @param sndTimeoutInMs send timeout in ms, -1 will wait forever (or until interrupt (e.g. via state change)), 0 will not wait (return immediately if cannot send)
/// @return Number of bytes that have been queued. -2 If queueing was not possible or timed out. -1 if there was an error.
int64_t Send(FairMQParts& parts, const std::string& chan, const int i = 0, int sndTimeoutInMs = -1) const
int64_t Send(FairMQParts& parts, const std::string& chan, const int i = 0, int sndTimeoutInMs = -1)
{
return fChannels.at(chan).at(i).Send(parts.fParts, sndTimeoutInMs);
}
@ -144,58 +144,58 @@ class FairMQDevice : public FairMQStateMachine
/// @param i channel index
/// @param rcvTimeoutInMs receive timeout in ms, -1 will wait forever (or until interrupt (e.g. via state change)), 0 will not wait (return immediately if cannot receive)
/// @return Number of bytes that have been received. -2 if reading from the queue was not possible or timed out. -1 if there was an error.
int64_t Receive(FairMQParts& parts, const std::string& chan, const int i = 0, int rcvTimeoutInMs = -1) const
int64_t Receive(FairMQParts& parts, const std::string& chan, const int i = 0, int rcvTimeoutInMs = -1)
{
return fChannels.at(chan).at(i).Receive(parts.fParts, rcvTimeoutInMs);
}
int64_t SendAsync(FairMQParts& parts, const std::string& chan, const int i = 0) const __attribute__((deprecated("For non-blocking Send, use timeout version with timeout of 0: Send(parts, \"channelA\", subchannelIndex, timeout);")))
int64_t SendAsync(FairMQParts& parts, const std::string& chan, const int i = 0) __attribute__((deprecated("For non-blocking Send, use timeout version with timeout of 0: Send(parts, \"channelA\", subchannelIndex, timeout);")))
{
return fChannels.at(chan).at(i).Send(parts.fParts, 0);
}
int64_t ReceiveAsync(FairMQParts& parts, const std::string& chan, const int i = 0) const __attribute__((deprecated("For non-blocking Receive, use timeout version with timeout of 0: Receive(parts, \"channelA\", subchannelIndex, timeout);")))
int64_t ReceiveAsync(FairMQParts& parts, const std::string& chan, const int i = 0) __attribute__((deprecated("For non-blocking Receive, use timeout version with timeout of 0: Receive(parts, \"channelA\", subchannelIndex, timeout);")))
{
return fChannels.at(chan).at(i).Receive(parts.fParts, 0);
}
/// @brief Getter for default transport factory
auto Transport() const -> const FairMQTransportFactory*
auto Transport() const -> FairMQTransportFactory*
{
return fTransportFactory.get();
}
template<typename... Args>
FairMQMessagePtr NewMessage(Args&&... args) const
FairMQMessagePtr NewMessage(Args&&... args)
{
return Transport()->CreateMessage(std::forward<Args>(args)...);
}
template<typename... Args>
FairMQMessagePtr NewMessageFor(const std::string& channel, int index, Args&&... args) const
FairMQMessagePtr NewMessageFor(const std::string& channel, int index, Args&&... args)
{
return fChannels.at(channel).at(index).NewMessage(std::forward<Args>(args)...);
}
template<typename T>
FairMQMessagePtr NewStaticMessage(const T& data) const
FairMQMessagePtr NewStaticMessage(const T& data)
{
return Transport()->NewStaticMessage(data);
}
template<typename T>
FairMQMessagePtr NewStaticMessageFor(const std::string& channel, int index, const T& data) const
FairMQMessagePtr NewStaticMessageFor(const std::string& channel, int index, const T& data)
{
return fChannels.at(channel).at(index).NewStaticMessage(data);
}
template<typename T>
FairMQMessagePtr NewSimpleMessage(const T& data) const
FairMQMessagePtr NewSimpleMessage(const T& data)
{
return Transport()->NewSimpleMessage(data);
}
template<typename T>
FairMQMessagePtr NewSimpleMessageFor(const std::string& channel, int index, const T& data) const
FairMQMessagePtr NewSimpleMessageFor(const std::string& channel, int index, const T& data)
{
return fChannels.at(channel).at(index).NewSimpleMessage(data);
}
@ -233,7 +233,7 @@ class FairMQDevice : public FairMQStateMachine
return fChannels.at(chans.at(0)).at(0).Transport()->CreatePoller(fChannels, chans);
}
FairMQPollerPtr NewPoller(const std::vector<const FairMQChannel*>& channels)
FairMQPollerPtr NewPoller(const std::vector<FairMQChannel*>& channels)
{
// if more than one channel provided, check compatibility
if (channels.size() > 1)
@ -490,8 +490,8 @@ class FairMQDevice : public FairMQStateMachine
void HandleMultipleTransportInput();
void PollForTransport(const FairMQTransportFactory* factory, const std::vector<std::string>& channelKeys);
bool HandleMsgInput(const std::string& chName, const InputMsgCallback& callback, int i) const;
bool HandleMultipartInput(const std::string& chName, const InputMultipartCallback& callback, int i) const;
bool HandleMsgInput(const std::string& chName, const InputMsgCallback& callback, int i);
bool HandleMultipartInput(const std::string& chName, const InputMultipartCallback& callback, int i);
void CreateOwnConfig();

8
fairmq/FairMQMessage.h
View File

@ -15,10 +15,13 @@
#include <fairmq/Transports.h>
using fairmq_free_fn = void(void* data, void* hint);
class FairMQTransportFactory;
class FairMQMessage
{
public:
FairMQMessage() = default;
FairMQMessage(FairMQTransportFactory* factory):fTransport{factory} {}
virtual void Rebuild() = 0;
virtual void Rebuild(const size_t size) = 0;
virtual void Rebuild(void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr) = 0;
@ -29,11 +32,16 @@ class FairMQMessage
virtual bool SetUsedSize(const size_t size) = 0;
virtual fair::mq::Transport GetType() const = 0;
FairMQTransportFactory* GetTransport() { return fTransport; }
//void SetTransport(FairMQTransportFactory* transport) { fTransport = transport; }
virtual void Copy(const std::unique_ptr<FairMQMessage>& msg) __attribute__((deprecated("Use 'Copy(const FairMQMessage& msg)'"))) = 0;
virtual void Copy(const FairMQMessage& msg) = 0;
virtual ~FairMQMessage() {};
private:
FairMQTransportFactory* fTransport{nullptr};
};
using FairMQMessagePtr = std::unique_ptr<FairMQMessage>;

20
fairmq/FairMQTransportFactory.h
View File

@ -46,20 +46,20 @@ class FairMQTransportFactory
/// @brief Create empty FairMQMessage
/// @return pointer to FairMQMessage
virtual FairMQMessagePtr CreateMessage() const = 0;
virtual FairMQMessagePtr CreateMessage() = 0;
/// @brief Create new FairMQMessage of specified size
/// @param size message size
/// @return pointer to FairMQMessage
virtual FairMQMessagePtr CreateMessage(const size_t size) const = 0;
virtual FairMQMessagePtr CreateMessage(const size_t size) = 0;
/// @brief Create new FairMQMessage with user provided buffer and size
/// @param data pointer to user provided buffer
/// @param size size of the user provided buffer
/// @param ffn callback, called when the message is transfered (and can be deleted)
/// @param obj optional helper pointer that can be used in the callback
/// @return pointer to FairMQMessage
virtual FairMQMessagePtr CreateMessage(void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr) const = 0;
virtual FairMQMessagePtr CreateMessage(void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr) = 0;
virtual FairMQMessagePtr CreateMessage(FairMQUnmanagedRegionPtr& unmanagedRegion, void* data, const size_t size, void* hint = 0) const = 0;
virtual FairMQMessagePtr CreateMessage(FairMQUnmanagedRegionPtr& unmanagedRegion, void* data, const size_t size, void* hint = 0) = 0;
/// Create a socket
virtual FairMQSocketPtr CreateSocket(const std::string& type, const std::string& name) const = 0;
@ -67,7 +67,7 @@ class FairMQTransportFactory
/// Create a poller for a single channel (all subchannels)
virtual FairMQPollerPtr CreatePoller(const std::vector<FairMQChannel>& channels) const = 0;
/// Create a poller for specific channels
virtual FairMQPollerPtr CreatePoller(const std::vector<const FairMQChannel*>& channels) const = 0;
virtual FairMQPollerPtr CreatePoller(const std::vector<FairMQChannel*>& channels) const = 0;
/// Create a poller for specific channels (all subchannels)
virtual FairMQPollerPtr CreatePoller(const std::unordered_map<std::string, std::vector<FairMQChannel>>& channelsMap, const std::vector<std::string>& channelList) const = 0;
@ -95,7 +95,7 @@ class FairMQTransportFactory
}
template<typename T>
FairMQMessagePtr NewSimpleMessage(const T& data) const
FairMQMessagePtr NewSimpleMessage(const T& data)
{
// todo: is_trivially_copyable not available on gcc < 5, workaround?
// static_assert(std::is_trivially_copyable<T>::value, "The argument type for NewSimpleMessage has to be trivially copyable!");
@ -104,13 +104,13 @@ class FairMQTransportFactory
}
template<std::size_t N>
FairMQMessagePtr NewSimpleMessage(const char(&data)[N]) const
FairMQMessagePtr NewSimpleMessage(const char(&data)[N])
{
std::string* msgStr = new std::string(data);
return CreateMessage(const_cast<char*>(msgStr->c_str()), msgStr->length(), FairMQSimpleMsgCleanup<std::string>, msgStr);
}
FairMQMessagePtr NewSimpleMessage(const std::string& str) const
FairMQMessagePtr NewSimpleMessage(const std::string& str)
{
std::string* msgStr = new std::string(str);
@ -118,12 +118,12 @@ class FairMQTransportFactory
}
template<typename T>
FairMQMessagePtr NewStaticMessage(const T& data) const
FairMQMessagePtr NewStaticMessage(const T& data)
{
return CreateMessage(data, sizeof(T), FairMQNoCleanup, nullptr);
}
FairMQMessagePtr NewStaticMessage(const std::string& str) const
FairMQMessagePtr NewStaticMessage(const std::string& str)
{
return CreateMessage(const_cast<char*>(str.c_str()), str.length(), FairMQNoCleanup, nullptr);
}

10
fairmq/MemoryResourceTools.h
View File

@ -75,17 +75,15 @@ std::vector<const ElemT, boost::container::pmr::polymorphic_allocator<const Elem
};
//_________________________________________________________________________________________________
/// Return a vector of const ElemT, takes ownership of the message, needs an
/// upstream global
/// ChannelResource to register the message.
/// Return a vector of const ElemT, takes ownership of the message
template<typename ElemT>
std::vector<const ElemT, OwningMessageSpectatorAllocator<const ElemT>>
adoptVector(size_t nelem, ChannelResource *upstream, FairMQMessagePtr message)
adoptVector(size_t nelem, FairMQMessagePtr message)
{
return std::vector<const ElemT, OwningMessageSpectatorAllocator<const ElemT>>(
nelem,
OwningMessageSpectatorAllocator<const ElemT>(
MessageResource{std::move(message), upstream}));
MessageResource{std::move(message)}));
};
//_________________________________________________________________________________________________
@ -93,7 +91,7 @@ std::vector<const ElemT, OwningMessageSpectatorAllocator<const ElemT>>
// This returns a unique_ptr of const vector, does not allow modifications at
// the cost of pointer
// semantics for access.
// use auto or decltype to catch the return value (or use span)
// use auto or decltype to catch the return type.
// template<typename ElemT>
// auto adoptVector(size_t nelem, FairMQMessage* message)
//{

15
fairmq/MemoryResources.cxx
View File

@ -17,9 +17,14 @@
void *fair::mq::ChannelResource::do_allocate(std::size_t bytes, std::size_t /*alignment*/)
{
FairMQMessagePtr message;
message = factory->CreateMessage(bytes);
void *addr = message->GetData();
messageMap[addr] = std::move(message);
return addr;
return setMessage(factory->CreateMessage(bytes));
};
fair::mq::MessageResource::MessageResource(FairMQMessagePtr message)
: mUpstream{message->GetTransport()->GetMemoryResource()}
, mMessageSize{message->GetSize()}
, mMessageData{mUpstream ? mUpstream->setMessage(std::move(message))
: throw std::runtime_error(
"MessageResource::MessageResource message has no upstream resource set")}
{
}

21
fairmq/MemoryResources.h
View File

@ -47,7 +47,7 @@ class FairMQMemoryResource : public boost::container::pmr::memory_resource
/// a message does not make sense!
virtual FairMQMessagePtr getMessage(void *p) = 0;
virtual void *setMessage(FairMQMessagePtr) = 0;
virtual const FairMQTransportFactory *getTransportFactory() const noexcept = 0;
virtual FairMQTransportFactory *getTransportFactory() noexcept = 0;
virtual size_t getNumberOfMessages() const noexcept = 0;
};
@ -59,7 +59,7 @@ class FairMQMemoryResource : public boost::container::pmr::memory_resource
class ChannelResource : public FairMQMemoryResource
{
protected:
const FairMQTransportFactory *factory{nullptr};
FairMQTransportFactory *factory{nullptr};
// TODO: for now a map to keep track of allocations, something else would
// probably be
// faster, but for now this does not need to be fast.
@ -68,7 +68,7 @@ class ChannelResource : public FairMQMemoryResource
public:
ChannelResource() = delete;
ChannelResource(const FairMQTransportFactory *_factory)
ChannelResource(FairMQTransportFactory *_factory)
: FairMQMemoryResource()
, factory(_factory)
, messageMap()
@ -92,7 +92,7 @@ class ChannelResource : public FairMQMemoryResource
return addr;
}
const FairMQTransportFactory *getTransportFactory() const noexcept override { return factory; }
FairMQTransportFactory *getTransportFactory() noexcept override { return factory; }
size_t getNumberOfMessages() const noexcept override { return messageMap.size(); }
@ -125,7 +125,7 @@ class SpectatorMessageResource : public FairMQMemoryResource
}
FairMQMessagePtr getMessage(void * /*p*/) override { return nullptr; }
const FairMQTransportFactory *getTransportFactory() const noexcept override { return nullptr; }
FairMQTransportFactory *getTransportFactory() noexcept override { return nullptr; }
size_t getNumberOfMessages() const noexcept override { return 0; }
void *setMessage(FairMQMessagePtr) override { return nullptr; }
@ -180,14 +180,7 @@ class MessageResource : public FairMQMemoryResource
MessageResource &operator=(const MessageResource &) = default;
MessageResource &operator=(MessageResource &&) = default;
MessageResource(FairMQMessagePtr message, FairMQMemoryResource *upstream)
: mUpstream{upstream}
, mMessageSize{message->GetSize()}
, mMessageData{mUpstream ? mUpstream->setMessage(std::move(message))
: throw std::runtime_error(
"MessageResource::MessageResource upstream is nullptr")}
{
}
MessageResource(FairMQMessagePtr message);
FairMQMessagePtr getMessage(void *p) override { return mUpstream->getMessage(p); }
void *setMessage(FairMQMessagePtr message) override
@ -195,7 +188,7 @@ class MessageResource : public FairMQMemoryResource
return mUpstream->setMessage(std::move(message));
}
const FairMQTransportFactory *getTransportFactory() const noexcept override { return nullptr; }
FairMQTransportFactory *getTransportFactory() noexcept override { return nullptr; }
size_t getNumberOfMessages() const noexcept override { return mMessageData ? 1 : 0; }
protected:

2
fairmq/devices/FairMQMerger.cxx
View File

@ -49,7 +49,7 @@ void FairMQMerger::Run()
{
int numInputs = fChannels.at(fInChannelName).size();
vector<const FairMQChannel*> chans;
vector<FairMQChannel*> chans;
for (auto& chan : fChannels.at(fInChannelName))
{

20
fairmq/nanomsg/FairMQMessageNN.cxx
View File

@ -24,8 +24,9 @@ using namespace std;
fair::mq::Transport FairMQMessageNN::fTransportType = fair::mq::Transport::NN;
FairMQMessageNN::FairMQMessageNN()
: fMessage(nullptr)
FairMQMessageNN::FairMQMessageNN(FairMQTransportFactory* factory)
: FairMQMessage{factory}
, fMessage(nullptr)
, fSize(0)
, fHint(0)
, fReceiving(false)
@ -38,8 +39,9 @@ FairMQMessageNN::FairMQMessageNN()
}
}
FairMQMessageNN::FairMQMessageNN(const size_t size)
: fMessage(nullptr)
FairMQMessageNN::FairMQMessageNN(const size_t size, FairMQTransportFactory* factory)
: FairMQMessage{factory}
, fMessage(nullptr)
, fSize(0)
, fHint(0)
, fReceiving(false)
@ -59,8 +61,9 @@ FairMQMessageNN::FairMQMessageNN(const size_t size)
* create FairMQMessage object only with size parameter and fill it with data.
* possible TODO: make this zero copy (will should then be as efficient as ZeroMQ).
*/
FairMQMessageNN::FairMQMessageNN(void* data, const size_t size, fairmq_free_fn* ffn, void* hint)
: fMessage(nullptr)
FairMQMessageNN::FairMQMessageNN(void* data, const size_t size, fairmq_free_fn* ffn, void* hint, FairMQTransportFactory* factory)
: FairMQMessage{factory}
, fMessage(nullptr)
, fSize(0)
, fHint(0)
, fReceiving(false)
@ -86,8 +89,9 @@ FairMQMessageNN::FairMQMessageNN(void* data, const size_t size, fairmq_free_fn*
}
}
FairMQMessageNN::FairMQMessageNN(FairMQUnmanagedRegionPtr& region, void* data, const size_t size, void* hint)
: fMessage(data)
FairMQMessageNN::FairMQMessageNN(FairMQUnmanagedRegionPtr& region, void* /*data*/, const size_t size, void* hint, FairMQTransportFactory* factory)
: FairMQMessage{factory}
, fMessage(nullptr)
, fSize(size)
, fHint(reinterpret_cast<size_t>(hint))
, fReceiving(false)

8
fairmq/nanomsg/FairMQMessageNN.h
View File

@ -29,10 +29,10 @@ class FairMQMessageNN final : public FairMQMessage
friend class FairMQSocketNN;
public:
FairMQMessageNN();
FairMQMessageNN(const size_t size);
FairMQMessageNN(void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr);
FairMQMessageNN(FairMQUnmanagedRegionPtr& region, void* data, const size_t size, void* hint = 0);
FairMQMessageNN(FairMQTransportFactory* factory = nullptr);
FairMQMessageNN(const size_t size, FairMQTransportFactory* factory = nullptr);
FairMQMessageNN(void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr, FairMQTransportFactory* factory = nullptr);
FairMQMessageNN(FairMQUnmanagedRegionPtr& region, void* data, const size_t size, void* hint = 0, FairMQTransportFactory* factory = nullptr);
FairMQMessageNN(const FairMQMessageNN&) = delete;
FairMQMessageNN operator=(const FairMQMessageNN&) = delete;

2
fairmq/nanomsg/FairMQPollerNN.cxx
View File

@ -44,7 +44,7 @@ FairMQPollerNN::FairMQPollerNN(const vector<FairMQChannel>& channels)
}
}
FairMQPollerNN::FairMQPollerNN(const vector<const FairMQChannel*>& channels)
FairMQPollerNN::FairMQPollerNN(const vector<FairMQChannel*>& channels)
: fItems()
, fNumItems(0)
, fOffsetMap()

4
fairmq/nanomsg/FairMQPollerNN.h
View File

@ -33,7 +33,7 @@ class FairMQPollerNN final : public FairMQPoller
public:
FairMQPollerNN(const std::vector<FairMQChannel>& channels);
FairMQPollerNN(const std::vector<const FairMQChannel*>& channels);
FairMQPollerNN(const std::vector<FairMQChannel*>& channels);
FairMQPollerNN(const std::unordered_map<std::string, std::vector<FairMQChannel>>& channelsMap, const std::vector<std::string>& channelList);
FairMQPollerNN(const FairMQPollerNN&) = delete;
@ -56,4 +56,4 @@ class FairMQPollerNN final : public FairMQPoller
std::unordered_map<std::string, int> fOffsetMap;
};
#endif /* FAIRMQPOLLERNN_H_ */
#endif /* FAIRMQPOLLERNN_H_ */

18
fairmq/nanomsg/FairMQTransportFactoryNN.cxx
View File

@ -23,24 +23,24 @@ FairMQTransportFactoryNN::FairMQTransportFactoryNN(const string& id, const FairM
LOG(debug) << "Transport: Using nanomsg library";
}
FairMQMessagePtr FairMQTransportFactoryNN::CreateMessage() const
FairMQMessagePtr FairMQTransportFactoryNN::CreateMessage()
{
return unique_ptr<FairMQMessage>(new FairMQMessageNN());
return unique_ptr<FairMQMessage>(new FairMQMessageNN(this));
}
FairMQMessagePtr FairMQTransportFactoryNN::CreateMessage(const size_t size) const
FairMQMessagePtr FairMQTransportFactoryNN::CreateMessage(const size_t size)
{
return unique_ptr<FairMQMessage>(new FairMQMessageNN(size));
return unique_ptr<FairMQMessage>(new FairMQMessageNN(size, this));
}
FairMQMessagePtr FairMQTransportFactoryNN::CreateMessage(void* data, const size_t size, fairmq_free_fn* ffn, void* hint) const
FairMQMessagePtr FairMQTransportFactoryNN::CreateMessage(void* data, const size_t size, fairmq_free_fn* ffn, void* hint)
{
return unique_ptr<FairMQMessage>(new FairMQMessageNN(data, size, ffn, hint));
return unique_ptr<FairMQMessage>(new FairMQMessageNN(data, size, ffn, hint, this));
}
FairMQMessagePtr FairMQTransportFactoryNN::CreateMessage(FairMQUnmanagedRegionPtr& region, void* data, const size_t size, void* hint) const
FairMQMessagePtr FairMQTransportFactoryNN::CreateMessage(FairMQUnmanagedRegionPtr& region, void* data, const size_t size, void* hint)
{
return unique_ptr<FairMQMessage>(new FairMQMessageNN(region, data, size, hint));
return unique_ptr<FairMQMessage>(new FairMQMessageNN(region, data, size, hint, this));
}
FairMQSocketPtr FairMQTransportFactoryNN::CreateSocket(const string& type, const string& name) const
@ -55,7 +55,7 @@ FairMQPollerPtr FairMQTransportFactoryNN::CreatePoller(const vector<FairMQChanne
return unique_ptr<FairMQPoller>(new FairMQPollerNN(channels));
}
FairMQPollerPtr FairMQTransportFactoryNN::CreatePoller(const std::vector<const FairMQChannel*>& channels) const
FairMQPollerPtr FairMQTransportFactoryNN::CreatePoller(const std::vector<FairMQChannel*>& channels) const
{
return unique_ptr<FairMQPoller>(new FairMQPollerNN(channels));
}

10
fairmq/nanomsg/FairMQTransportFactoryNN.h
View File

@ -25,15 +25,15 @@ class FairMQTransportFactoryNN final : public FairMQTransportFactory
FairMQTransportFactoryNN(const std::string& id = "", const FairMQProgOptions* config = nullptr);
~FairMQTransportFactoryNN() override;
FairMQMessagePtr CreateMessage() const override;
FairMQMessagePtr CreateMessage(const size_t size) const override;
FairMQMessagePtr CreateMessage(void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr) const override;
FairMQMessagePtr CreateMessage(FairMQUnmanagedRegionPtr& region, void* data, const size_t size, void* hint = 0) const override;
FairMQMessagePtr CreateMessage() override;
FairMQMessagePtr CreateMessage(const size_t size) override;
FairMQMessagePtr CreateMessage(void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr) override;
FairMQMessagePtr CreateMessage(FairMQUnmanagedRegionPtr& region, void* data, const size_t size, void* hint = 0) override;
FairMQSocketPtr CreateSocket(const std::string& type, const std::string& name) const override;
FairMQPollerPtr CreatePoller(const std::vector<FairMQChannel>& channels) const override;
FairMQPollerPtr CreatePoller(const std::vector<const FairMQChannel*>& channels) const override;
FairMQPollerPtr CreatePoller(const std::vector<FairMQChannel*>& channels) const override;
FairMQPollerPtr CreatePoller(const std::unordered_map<std::string, std::vector<FairMQChannel>>& channelsMap, const std::vector<std::string>& channelList) const override;
FairMQUnmanagedRegionPtr CreateUnmanagedRegion(const size_t size, FairMQRegionCallback callback) const override;

20
fairmq/shmem/FairMQMessageSHM.cxx
View File

@ -25,8 +25,9 @@ namespace bpt = ::boost::posix_time;
atomic<bool> FairMQMessageSHM::fInterrupted(false);
fair::mq::Transport FairMQMessageSHM::fTransportType = fair::mq::Transport::SHM;
FairMQMessageSHM::FairMQMessageSHM(Manager& manager)
: fManager(manager)
FairMQMessageSHM::FairMQMessageSHM(Manager& manager, FairMQTransportFactory* factory)
: FairMQMessage{factory}
, fManager(manager)
, fMessage()
, fQueued(false)
, fMetaCreated(false)
@ -44,8 +45,9 @@ FairMQMessageSHM::FairMQMessageSHM(Manager& manager)
fMetaCreated = true;
}
FairMQMessageSHM::FairMQMessageSHM(Manager& manager, const size_t size)
: fManager(manager)
FairMQMessageSHM::FairMQMessageSHM(Manager& manager, const size_t size, FairMQTransportFactory* factory)
: FairMQMessage{factory}
, fManager(manager)
, fMessage()
, fQueued(false)
, fMetaCreated(false)
@ -59,8 +61,9 @@ FairMQMessageSHM::FairMQMessageSHM(Manager& manager, const size_t size)
InitializeChunk(size);
}
FairMQMessageSHM::FairMQMessageSHM(Manager& manager, void* data, const size_t size, fairmq_free_fn* ffn, void* hint)
: fManager(manager)
FairMQMessageSHM::FairMQMessageSHM(Manager& manager, void* data, const size_t size, fairmq_free_fn* ffn, void* hint, FairMQTransportFactory* factory)
: FairMQMessage{factory}
, fManager(manager)
, fMessage()
, fQueued(false)
, fMetaCreated(false)
@ -85,8 +88,9 @@ FairMQMessageSHM::FairMQMessageSHM(Manager& manager, void* data, const size_t si
}
}
FairMQMessageSHM::FairMQMessageSHM(Manager& manager, FairMQUnmanagedRegionPtr& region, void* data, const size_t size, void* hint)
: fManager(manager)
FairMQMessageSHM::FairMQMessageSHM(Manager& manager, FairMQUnmanagedRegionPtr& region, void* data, const size_t size, void* hint, FairMQTransportFactory* factory)
: FairMQMessage{factory}
, fManager(manager)
, fMessage()
, fQueued(false)
, fMetaCreated(false)

8
fairmq/shmem/FairMQMessageSHM.h
View File

@ -27,10 +27,10 @@ class FairMQMessageSHM final : public FairMQMessage
friend class FairMQSocketSHM;
public:
FairMQMessageSHM(fair::mq::shmem::Manager& manager);
FairMQMessageSHM(fair::mq::shmem::Manager& manager, const size_t size);
FairMQMessageSHM(fair::mq::shmem::Manager& manager, void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr);
FairMQMessageSHM(fair::mq::shmem::Manager& manager, FairMQUnmanagedRegionPtr& region, void* data, const size_t size, void* hint = 0);
FairMQMessageSHM(fair::mq::shmem::Manager& manager, FairMQTransportFactory* factory = nullptr);
FairMQMessageSHM(fair::mq::shmem::Manager& manager, const size_t size, FairMQTransportFactory* factory = nullptr);
FairMQMessageSHM(fair::mq::shmem::Manager& manager, void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr, FairMQTransportFactory* factory = nullptr);
FairMQMessageSHM(fair::mq::shmem::Manager& manager, FairMQUnmanagedRegionPtr& region, void* data, const size_t size, void* hint = 0, FairMQTransportFactory* factory = nullptr);
FairMQMessageSHM(const FairMQMessageSHM&) = delete;
FairMQMessageSHM operator=(const FairMQMessageSHM&) = delete;

2
fairmq/shmem/FairMQPollerSHM.cxx
View File

@ -42,7 +42,7 @@ FairMQPollerSHM::FairMQPollerSHM(const vector<FairMQChannel>& channels)
}
}
FairMQPollerSHM::FairMQPollerSHM(const vector<const FairMQChannel*>& channels)
FairMQPollerSHM::FairMQPollerSHM(const vector<FairMQChannel*>& channels)
: fItems()
, fNumItems(0)
, fOffsetMap()

4
fairmq/shmem/FairMQPollerSHM.h
View File

@ -26,7 +26,7 @@ class FairMQPollerSHM final : public FairMQPoller
public:
FairMQPollerSHM(const std::vector<FairMQChannel>& channels);
FairMQPollerSHM(const std::vector<const FairMQChannel*>& channels);
FairMQPollerSHM(const std::vector<FairMQChannel*>& channels);
FairMQPollerSHM(const std::unordered_map<std::string, std::vector<FairMQChannel>>& channelsMap, const std::vector<std::string>& channelList);
FairMQPollerSHM(const FairMQPollerSHM&) = delete;
@ -49,4 +49,4 @@ class FairMQPollerSHM final : public FairMQPoller
std::unordered_map<std::string, int> fOffsetMap;
};
#endif /* FAIRMQPOLLERSHM_H_ */
#endif /* FAIRMQPOLLERSHM_H_ */

18
fairmq/shmem/FairMQTransportFactorySHM.cxx
View File

@ -213,24 +213,24 @@ void FairMQTransportFactorySHM::SendHeartbeats()
}
}
FairMQMessagePtr FairMQTransportFactorySHM::CreateMessage() const
FairMQMessagePtr FairMQTransportFactorySHM::CreateMessage()
{
return unique_ptr<FairMQMessage>(new FairMQMessageSHM(*fManager));
return unique_ptr<FairMQMessage>(new FairMQMessageSHM(*fManager, this));
}
FairMQMessagePtr FairMQTransportFactorySHM::CreateMessage(const size_t size) const
FairMQMessagePtr FairMQTransportFactorySHM::CreateMessage(const size_t size)
{
return unique_ptr<FairMQMessage>(new FairMQMessageSHM(*fManager, size));
return unique_ptr<FairMQMessage>(new FairMQMessageSHM(*fManager, size, this));
}
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)
{
return unique_ptr<FairMQMessage>(new FairMQMessageSHM(*fManager, data, size, ffn, hint));
return unique_ptr<FairMQMessage>(new FairMQMessageSHM(*fManager, data, size, ffn, hint, this));
}
FairMQMessagePtr FairMQTransportFactorySHM::CreateMessage(FairMQUnmanagedRegionPtr& region, void* data, const size_t size, void* hint) const
FairMQMessagePtr FairMQTransportFactorySHM::CreateMessage(FairMQUnmanagedRegionPtr& region, void* data, const size_t size, void* hint)
{
return unique_ptr<FairMQMessage>(new FairMQMessageSHM(*fManager, region, data, size, hint));
return unique_ptr<FairMQMessage>(new FairMQMessageSHM(*fManager, region, data, size, hint, this));
}
FairMQSocketPtr FairMQTransportFactorySHM::CreateSocket(const string& type, const string& name) const
@ -244,7 +244,7 @@ FairMQPollerPtr FairMQTransportFactorySHM::CreatePoller(const vector<FairMQChann
return unique_ptr<FairMQPoller>(new FairMQPollerSHM(channels));
}
FairMQPollerPtr FairMQTransportFactorySHM::CreatePoller(const vector<const FairMQChannel*>& channels) const
FairMQPollerPtr FairMQTransportFactorySHM::CreatePoller(const vector<FairMQChannel*>& channels) const
{
return unique_ptr<FairMQPoller>(new FairMQPollerSHM(channels));
}

10
fairmq/shmem/FairMQTransportFactorySHM.h
View File

@ -33,15 +33,15 @@ class FairMQTransportFactorySHM final : public FairMQTransportFactory
FairMQTransportFactorySHM(const FairMQTransportFactorySHM&) = delete;
FairMQTransportFactorySHM operator=(const FairMQTransportFactorySHM&) = delete;
FairMQMessagePtr CreateMessage() const override;
FairMQMessagePtr CreateMessage(const size_t size) const override;
FairMQMessagePtr CreateMessage(void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr) const override;
FairMQMessagePtr CreateMessage(FairMQUnmanagedRegionPtr& region, void* data, const size_t size, void* hint = 0) const override;
FairMQMessagePtr CreateMessage() override;
FairMQMessagePtr CreateMessage(const size_t size) override;
FairMQMessagePtr CreateMessage(void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr) override;
FairMQMessagePtr CreateMessage(FairMQUnmanagedRegionPtr& region, void* data, const size_t size, void* hint = 0) override;
FairMQSocketPtr CreateSocket(const std::string& type, const std::string& name) const override;
FairMQPollerPtr CreatePoller(const std::vector<FairMQChannel>& channels) const override;
FairMQPollerPtr CreatePoller(const std::vector<const FairMQChannel*>& channels) const override;
FairMQPollerPtr CreatePoller(const std::vector<FairMQChannel*>& channels) const override;
FairMQPollerPtr CreatePoller(const std::unordered_map<std::string, std::vector<FairMQChannel>>& channelsMap, const std::vector<std::string>& channelList) const override;
FairMQUnmanagedRegionPtr CreateUnmanagedRegion(const size_t size, FairMQRegionCallback callback = nullptr) const override;

21
fairmq/zeromq/FairMQMessageZMQ.cxx
View File

@ -17,6 +17,7 @@
#include "FairMQLogger.h"
#include <fairmq/Tools.h>
#include "FairMQUnmanagedRegionZMQ.h"
#include <FairMQTransportFactory.h>
#include <cstring>
@ -24,8 +25,9 @@ using namespace std;
fair::mq::Transport FairMQMessageZMQ::fTransportType = fair::mq::Transport::ZMQ;
FairMQMessageZMQ::FairMQMessageZMQ()
: fUsedSizeModified(false)
FairMQMessageZMQ::FairMQMessageZMQ(FairMQTransportFactory* factory)
: FairMQMessage{factory}
, fUsedSizeModified(false)
, fUsedSize()
, fMsg(fair::mq::tools::make_unique<zmq_msg_t>())
, fViewMsg(nullptr)
@ -36,8 +38,9 @@ FairMQMessageZMQ::FairMQMessageZMQ()
}
}
FairMQMessageZMQ::FairMQMessageZMQ(const size_t size)
: fUsedSizeModified(false)
FairMQMessageZMQ::FairMQMessageZMQ(const size_t size, FairMQTransportFactory* factory)
: FairMQMessage{factory}
, fUsedSizeModified(false)
, fUsedSize(size)
, fMsg(fair::mq::tools::make_unique<zmq_msg_t>())
, fViewMsg(nullptr)
@ -48,8 +51,9 @@ FairMQMessageZMQ::FairMQMessageZMQ(const size_t size)
}
}
FairMQMessageZMQ::FairMQMessageZMQ(void* data, const size_t size, fairmq_free_fn* ffn, void* hint)
: fUsedSizeModified(false)
FairMQMessageZMQ::FairMQMessageZMQ(void* data, const size_t size, fairmq_free_fn* ffn, void* hint, FairMQTransportFactory* factory)
: FairMQMessage{factory}
, fUsedSizeModified(false)
, fUsedSize()
, fMsg(fair::mq::tools::make_unique<zmq_msg_t>())
, fViewMsg(nullptr)
@ -60,8 +64,9 @@ FairMQMessageZMQ::FairMQMessageZMQ(void* data, const size_t size, fairmq_free_fn
}
}
FairMQMessageZMQ::FairMQMessageZMQ(FairMQUnmanagedRegionPtr& region, void* data, const size_t size, void* hint)
: fUsedSizeModified(false)
FairMQMessageZMQ::FairMQMessageZMQ(FairMQUnmanagedRegionPtr& region, void* data, const size_t size, void* hint, FairMQTransportFactory* factory)
: FairMQMessage{factory}
, fUsedSizeModified(false)
, fUsedSize()
, fMsg(fair::mq::tools::make_unique<zmq_msg_t>())
, fViewMsg(nullptr)

9
fairmq/zeromq/FairMQMessageZMQ.h
View File

@ -23,6 +23,7 @@
#include "FairMQMessage.h"
#include "FairMQUnmanagedRegion.h"
class FairMQTransportFactory;
class FairMQSocketZMQ;
@ -31,10 +32,10 @@ class FairMQMessageZMQ final : public FairMQMessage
friend class FairMQSocketZMQ;
public:
FairMQMessageZMQ();
FairMQMessageZMQ(const size_t size);
FairMQMessageZMQ(void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr);
FairMQMessageZMQ(FairMQUnmanagedRegionPtr& region, void* data, const size_t size, void* hint = 0);
FairMQMessageZMQ(FairMQTransportFactory* = nullptr);
FairMQMessageZMQ(const size_t size, FairMQTransportFactory* = nullptr);
FairMQMessageZMQ(void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr, FairMQTransportFactory* = nullptr);
FairMQMessageZMQ(FairMQUnmanagedRegionPtr& region, void* data, const size_t size, void* hint = 0, FairMQTransportFactory* = nullptr);
void Rebuild() override;
void Rebuild(const size_t size) override;

2
fairmq/zeromq/FairMQPollerZMQ.cxx
View File

@ -43,7 +43,7 @@ FairMQPollerZMQ::FairMQPollerZMQ(const vector<FairMQChannel>& channels)
}
FairMQPollerZMQ::FairMQPollerZMQ(const std::vector<const FairMQChannel*>& channels)
FairMQPollerZMQ::FairMQPollerZMQ(const std::vector<FairMQChannel*>& channels)
: fItems()
, fNumItems(0)
, fOffsetMap()

4
fairmq/zeromq/FairMQPollerZMQ.h
View File

@ -34,7 +34,7 @@ class FairMQPollerZMQ final : public FairMQPoller
public:
FairMQPollerZMQ(const std::vector<FairMQChannel>& channels);
FairMQPollerZMQ(const std::vector<const FairMQChannel*>& channels);
FairMQPollerZMQ(const std::vector<FairMQChannel*>& channels);
FairMQPollerZMQ(const std::unordered_map<std::string, std::vector<FairMQChannel>>& channelsMap, const std::vector<std::string>& channelList);
FairMQPollerZMQ(const FairMQPollerZMQ&) = delete;
@ -57,4 +57,4 @@ class FairMQPollerZMQ final : public FairMQPoller
std::unordered_map<std::string, int> fOffsetMap;
};
#endif /* FAIRMQPOLLERZMQ_H_ */
#endif /* FAIRMQPOLLERZMQ_H_ */

18
fairmq/zeromq/FairMQTransportFactoryZMQ.cxx
View File

@ -49,24 +49,24 @@ FairMQTransportFactoryZMQ::FairMQTransportFactoryZMQ(const string& id, const Fai
}
FairMQMessagePtr FairMQTransportFactoryZMQ::CreateMessage() const
FairMQMessagePtr FairMQTransportFactoryZMQ::CreateMessage()
{
return unique_ptr<FairMQMessage>(new FairMQMessageZMQ());
return unique_ptr<FairMQMessage>(new FairMQMessageZMQ(this));
}
FairMQMessagePtr FairMQTransportFactoryZMQ::CreateMessage(const size_t size) const
FairMQMessagePtr FairMQTransportFactoryZMQ::CreateMessage(const size_t size)
{
return unique_ptr<FairMQMessage>(new FairMQMessageZMQ(size));
return unique_ptr<FairMQMessage>(new FairMQMessageZMQ(size, this));
}
FairMQMessagePtr FairMQTransportFactoryZMQ::CreateMessage(void* data, const size_t size, fairmq_free_fn* ffn, void* hint) const
FairMQMessagePtr FairMQTransportFactoryZMQ::CreateMessage(void* data, const size_t size, fairmq_free_fn* ffn, void* hint)
{
return unique_ptr<FairMQMessage>(new FairMQMessageZMQ(data, size, ffn, hint));
return unique_ptr<FairMQMessage>(new FairMQMessageZMQ(data, size, ffn, hint, this));
}
FairMQMessagePtr FairMQTransportFactoryZMQ::CreateMessage(FairMQUnmanagedRegionPtr& region, void* data, const size_t size, void* hint) const
FairMQMessagePtr FairMQTransportFactoryZMQ::CreateMessage(FairMQUnmanagedRegionPtr& region, void* data, const size_t size, void* hint)
{
return unique_ptr<FairMQMessage>(new FairMQMessageZMQ(region, data, size, hint));
return unique_ptr<FairMQMessage>(new FairMQMessageZMQ(region, data, size, hint, this));
}
FairMQSocketPtr FairMQTransportFactoryZMQ::CreateSocket(const string& type, const string& name) const
@ -80,7 +80,7 @@ FairMQPollerPtr FairMQTransportFactoryZMQ::CreatePoller(const vector<FairMQChann
return unique_ptr<FairMQPoller>(new FairMQPollerZMQ(channels));
}
FairMQPollerPtr FairMQTransportFactoryZMQ::CreatePoller(const std::vector<const FairMQChannel*>& channels) const
FairMQPollerPtr FairMQTransportFactoryZMQ::CreatePoller(const std::vector<FairMQChannel*>& channels) const
{
return unique_ptr<FairMQPoller>(new FairMQPollerZMQ(channels));
}

10
fairmq/zeromq/FairMQTransportFactoryZMQ.h
View File

@ -34,15 +34,15 @@ class FairMQTransportFactoryZMQ final : public FairMQTransportFactory
~FairMQTransportFactoryZMQ() override;
FairMQMessagePtr CreateMessage() const override;
FairMQMessagePtr CreateMessage(const size_t size) const override;
FairMQMessagePtr CreateMessage(void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr) const override;
FairMQMessagePtr CreateMessage(FairMQUnmanagedRegionPtr& region, void* data, const size_t size, void* hint = 0) const override;
FairMQMessagePtr CreateMessage() override;
FairMQMessagePtr CreateMessage(const size_t size) override;
FairMQMessagePtr CreateMessage(void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr) override;
FairMQMessagePtr CreateMessage(FairMQUnmanagedRegionPtr& region, void* data, const size_t size, void* hint = 0) override;
FairMQSocketPtr CreateSocket(const std::string& type, const std::string& name) const override;
FairMQPollerPtr CreatePoller(const std::vector<FairMQChannel>& channels) const override;
FairMQPollerPtr CreatePoller(const std::vector<const FairMQChannel*>& channels) const override;
FairMQPollerPtr CreatePoller(const std::vector<FairMQChannel*>& channels) const override;
FairMQPollerPtr CreatePoller(const std::unordered_map<std::string, std::vector<FairMQChannel>>& channelsMap, const std::vector<std::string>& channelList) const override;
FairMQUnmanagedRegionPtr CreateUnmanagedRegion(const size_t size, FairMQRegionCallback callback) const override;

2
test/helper/devices/TestPollIn.h
View File

@ -43,7 +43,7 @@ class PollIn : public FairMQDevice
auto Run() -> void override
{
vector<const FairMQChannel*> chans;
vector<FairMQChannel*> chans;
chans.push_back(&fChannels.at("data1").at(0));
chans.push_back(&fChannels.at("data2").at(0));

2
test/memory_resources/_memory_resources.cxx
View File

@ -166,7 +166,7 @@ TEST(MemoryResources, adoptVector_test)
std::memcpy(message->GetData(), tmpBuf, 3 * sizeof(testData));
auto adoptedOwner =
adoptVector<testData>(3, factoryZMQ->GetMemoryResource(), std::move(message));
adoptVector<testData>(3, std::move(message));
EXPECT_TRUE(adoptedOwner[0].i == 3);
EXPECT_TRUE(adoptedOwner[1].i == 2);
EXPECT_TRUE(adoptedOwner[2].i == 1);